def test_add_flow_without_icmp_code_and_icmp_type(self):
        """Test plugin deny add flow without icmp-code and icmp-type."""

        data = {
            "kind": "Access Control List",
            "rules": [{
                "id": 1,
                "protocol": "icmp",
                "source": "10.0.0.1/32",
                "destination": "10.0.0.2/32",
                "icmp-options": {
                }
            }]
        }

        rule = dumps(data['rules'][0], sort_keys=True)

        assert_raises_regexp(
            ValueError,
            "Error building ACL Json. Malformed input data: \n"
            "Missing icmp-code or icmp-type icmp options:\n%s" %
            rule,
            self.odl.add_flow,
            data
        )

    def test_add_flow_with_only_source(self):
        """Test plugin deny add flow with only source."""

        data = {
            "kind": "Access Control List",
            "rules": [{
                "action": "permit",
                "description": "Restrict environment",
                "icmp-options": {
                    "icmp-code": "0",
                    "icmp-type": "8"
                },
                "id": "82325",
                "owner": "networkapi",
                "protocol": "icmp",
                "source": "0.0.0.0/0"
            }]
        }

        rule = dumps(data['rules'][0], sort_keys=True)

        assert_raises_regexp(
            ValueError,
            "Error building ACL Json. Malformed input data: \n%s" %
            rule,
            self.odl.add_flow,
            data
        )

def test_verb_alias_config_initialization():
    cwd = os.getcwd()
    test_folder = os.path.join(cwd, 'test')
    # Test target directory does not exist failure
    with assert_raises_regexp(RuntimeError, "Cannot initialize verb aliases because catkin configuration path"):
        config.initialize_verb_aliases(test_folder)
    # Test normal case
    os.makedirs(test_folder)
    config.initialize_verb_aliases(test_folder)
    assert os.path.isdir(test_folder)
    assert os.path.isdir(os.path.join(test_folder, 'verb_aliases'))
    defaults_path = os.path.join(test_folder, 'verb_aliases', '00-default-aliases.yaml')
    assert os.path.isfile(defaults_path)
    # Assert a second invocation is fine
    config.initialize_verb_aliases(test_folder)
    # Check that replacement of defaults works
    with open(defaults_path, 'w') as f:
        f.write("This should be overwritten (simulation of update needed)")
    with redirected_stdio() as (out, err):
        config.initialize_verb_aliases(test_folder)
    assert "Warning, builtin verb aliases at" in out.getvalue(), out.getvalue()
    shutil.rmtree(test_folder)
    # Check failure from verb aliases folder existing as a file
    os.makedirs(test_folder)
    with open(os.path.join(test_folder, 'verb_aliases'), 'w') as f:
        f.write("this will cause a RuntimeError")
    with assert_raises_regexp(RuntimeError, "The catkin verb aliases config directory"):
        config.initialize_verb_aliases(test_folder)
    shutil.rmtree(test_folder)

def test_convolutional_sequence_with_no_input_size():
    # suppose x is outputted by some RNN
    x = tensor.tensor4('x')
    filter_size = (1, 1)
    num_filters = 2
    num_channels = 1
    pooling_size = (1, 1)
    conv = Convolutional(filter_size, num_filters, tied_biases=False,
                         weights_init=Constant(1.), biases_init=Constant(1.))
    act = Rectifier()
    pool = MaxPooling(pooling_size)

    bad_seq = ConvolutionalSequence([conv, act, pool], num_channels,
                                    tied_biases=False)
    assert_raises_regexp(ValueError, 'Cannot infer bias size \S+',
                         bad_seq.initialize)

    seq = ConvolutionalSequence([conv, act, pool], num_channels,
                                tied_biases=True)
    try:
        seq.initialize()
        out = seq.apply(x)
    except TypeError:
        assert False, "This should have succeeded"

    assert out.ndim == 4

    def test_create_external_integration(self):
        # A newly created Collection has no associated ExternalIntegration.
        collection, ignore = get_one_or_create(
            self._db, Collection, name=self._str
        )
        eq_(None, collection.external_integration_id)
        assert_raises_regexp(
            ValueError,
            "No known external integration for collection",
            getattr, collection, 'external_integration'
        )

        # We can create one with create_external_integration().
        overdrive = ExternalIntegration.OVERDRIVE
        integration = collection.create_external_integration(protocol=overdrive)
        eq_(integration.id, collection.external_integration_id)
        eq_(overdrive, integration.protocol)

        # If we call create_external_integration() again we get the same
        # ExternalIntegration as before.
        integration2 = collection.create_external_integration(protocol=overdrive)
        eq_(integration, integration2)


        # If we try to initialize an ExternalIntegration with a different
        # protocol, we get an error.
        assert_raises_regexp(
            ValueError,
            "Located ExternalIntegration, but its protocol \(Overdrive\) does not match desired protocol \(blah\).",
            collection.create_external_integration,
            protocol="blah"
        )

    def test_lane_loading(self):
        # The default setup loads lane IDs properly.
        gate = COPPAGate(self._default_library, self.integration)
        eq_(self.lane1.id, gate.yes_lane_id)
        eq_(self.lane2.id, gate.no_lane_id)

        # If a lane isn't associated with the right library, the
        # COPPAGate is misconfigured and cannot be instantiated.
        library = self._library()
        self.lane1.library = library
        self._db.commit()
        assert_raises_regexp(
            CannotLoadConfiguration,
            "Lane .* is for the wrong library",
            COPPAGate,
            self._default_library, self.integration
        )
        self.lane1.library_id = self._default_library.id

        # If the lane ID doesn't correspond to a real lane, the
        # COPPAGate cannot be instantiated.
        ConfigurationSetting.for_library_and_externalintegration(
            self._db, COPPAGate.REQUIREMENT_MET_LANE, self._default_library,
            self.integration
        ).value = -100
        assert_raises_regexp(
            CannotLoadConfiguration, "No lane with ID: -100",
            COPPAGate, self._default_library, self.integration
        )

 def test_missing_error_code(self):
     data = self.sample_data("missing_error_code.xml")
     parser = HoldReleaseResponseParser()
     assert_raises_regexp(
         RemoteInitiatedServerError, "No status code!", 
         parser.process_all, data
     )

def error_is_raised_if_too_many_positional_arguments_are_passed_to_init():
    User = dodge.data_class("User", ["username", "password"])
    
    assert_raises_regexp(
        TypeError, r"takes 2 positional arguments but 3 were given",
        lambda: User("bob", "password1", "salty")
    )

 def test_internal_server_error(self):
     data = self.sample_data("internal_server_error.xml")
     parser = HoldReleaseResponseParser()
     assert_raises_regexp(
         RemoteInitiatedServerError, "Internal Server Error", 
         parser.process_all, data
     )

 def test_bad_connection_remote_pin_test(self):
     api = self.mock_api(bad_connection=True)
     assert_raises_regexp(
         RemoteInitiatedServerError,
         "Could not connect!",
         api.remote_pin_test, "key", "pin"
     )

 def test_broken_service_remote_pin_test(self):
     api = self.mock_api(failure_status_code=502)
     assert_raises_regexp(
         RemoteInitiatedServerError,
         "Got unexpected response code 502. Content: Error 502",
         api.remote_pin_test, "key", "pin"
     )

    def test_simple(self):
        p = SimpleAuthenticationProvider
        integration = self._external_integration(self._str)

        assert_raises_regexp(
            CannotLoadConfiguration,
            "Test identifier and password not set.",
            p, self._default_library, integration
        )

        integration.setting(p.TEST_IDENTIFIER).value = "barcode"
        integration.setting(p.TEST_PASSWORD).value = "pass"
        provider = p(self._default_library, integration)

        eq_(None, provider.remote_authenticate("user", "wrongpass"))
        eq_(None, provider.remote_authenticate("user", None))
        eq_(None, provider.remote_authenticate(None, "pass"))
        user = provider.remote_authenticate("barcode", "pass")
        assert isinstance(user, PatronData)
        eq_("barcode", user.authorization_identifier)
        eq_("barcode_id", user.permanent_id)
        eq_("barcode_username", user.username)

        # User can also authenticate by their 'username'
        user2 = provider.remote_authenticate("barcode_username", "pass")
        eq_("barcode", user2.authorization_identifier)

    def test_for_collection(self):
        # This collection has no mirror_integration, so
        # there is no MirrorUploader for it.
        collection = self._collection()
        eq_(None, MirrorUploader.for_collection(collection))

        # We can tell the method that we're okay with a sitewide
        # integration instead of an integration specifically for this
        # collection.
        sitewide_integration = self._integration
        uploader = MirrorUploader.for_collection(collection, use_sitewide=True)
        assert isinstance(uploader, MirrorUploader)

        # This collection has a properly configured mirror_integration,
        # so it can have an MirrorUploader.
        collection.mirror_integration = self._integration
        uploader = MirrorUploader.for_collection(collection)
        assert isinstance(uploader, MirrorUploader)

        # This collection has a mirror_integration but it has the
        # wrong goal, so attempting to make an MirrorUploader for it
        # raises an exception.
        collection.mirror_integration.goal = ExternalIntegration.LICENSE_GOAL
        assert_raises_regexp(
            CannotLoadConfiguration,
            "from an integration with goal=licenses",
            MirrorUploader.for_collection, collection
        )

    def test_protocol_enforcement(self):
        """A CollectionMonitor can require that it be instantiated
        with a Collection that implements a certain protocol.
        """
        class NoProtocolMonitor(CollectionMonitor):
            SERVICE_NAME = "Test Monitor 1"
            PROTOCOL = None

        class OverdriveMonitor(CollectionMonitor):
            SERVICE_NAME = "Test Monitor 2"
            PROTOCOL = ExternalIntegration.OVERDRIVE

        # Two collections.
        c1 = self._collection(protocol=ExternalIntegration.OVERDRIVE)
        c2 = self._collection(protocol=ExternalIntegration.BIBLIOTHECA)

        # The NoProtocolMonitor can be instantiated with either one,
        # or with no Collection at all.
        NoProtocolMonitor(self._db, c1)
        NoProtocolMonitor(self._db, c2)
        NoProtocolMonitor(self._db, None)

        # The OverdriveMonitor can only be instantiated with the first one.
        OverdriveMonitor(self._db, c1)
        assert_raises_regexp(
            ValueError,
            "Collection protocol \(Bibliotheca\) does not match Monitor protocol \(Overdrive\)",
            OverdriveMonitor, self._db, c2
        )
        assert_raises(
            CollectionMissing,
            OverdriveMonitor, self._db, None
        )

def error_is_raised_if_init_value_is_missing():
    User = dodge.data_class("User", ["username", "password"])
    
    assert_raises_regexp(
        TypeError, "^Missing argument: 'password'$",
        lambda: User("bob")
    )

 def test_misconfigured_authentication_mode(self):
     assert_raises_regexp(
         CannotLoadConfiguration,
         "Unrecognized Millenium Patron API authentication mode: nosuchauthmode.",
         self.mock_api,
         auth_mode = 'nosuchauthmode'
     )

 def test_for_foreign_id_rejects_invalid_identifiers(self):
     assert_raises_regexp(
         ValueError,
         '"foo/bar" is not a valid Bibliotheca ID.',
         Identifier.for_foreign_id,
         self._db, Identifier.BIBLIOTHECA_ID, "foo/bar"
     )

 def test_identify_library_by_url(self):
     assert_raises_regexp(
         Exception,
         "Could not locate library with URL http://bar/. Available URLs: http://foo/",
         self.script.set_secret,
         self._db, "http://bar/", "vendorid", "libraryname", "secret", None
     )

    def test_by_name_and_protocol(self):
        name = "A name"
        protocol = ExternalIntegration.OVERDRIVE
        key = (name, protocol)

        # Cache is empty.
        eq_(HasFullTableCache.RESET, Collection._cache)

        collection1, is_new = Collection.by_name_and_protocol(
            self._db, name, ExternalIntegration.OVERDRIVE
        )
        eq_(True, is_new)

        # Cache was populated and then reset because we created a new
        # Collection.
        eq_(HasFullTableCache.RESET, Collection._cache)

        collection2, is_new = Collection.by_name_and_protocol(
            self._db, name, ExternalIntegration.OVERDRIVE
        )
        eq_(collection1, collection2)
        eq_(False, is_new)

        # This time the cache was not reset after being populated.
        eq_(collection1, Collection._cache[key])

        # You'll get an exception if you look up an existing name
        # but the protocol doesn't match.
        assert_raises_regexp(
            ValueError,
            'Collection "A name" does not use protocol "Bibliotheca".',
            Collection.by_name_and_protocol,
            self._db, name, ExternalIntegration.BIBLIOTHECA
        )

def test_untied_biases():
    x = tensor.tensor4('x')
    num_channels = 4
    num_filters = 3
    batch_size = 5
    filter_size = (3, 3)
    conv = Convolutional(filter_size, num_filters, num_channels,
                         weights_init=Constant(1.), biases_init=Constant(2.),
                         image_size=(28, 30), tied_biases=False)
    conv.initialize()

    y = conv.apply(x)
    func = function([x], y)

    # Untied biases provide a bias for every individual output
    assert_allclose(conv.b.eval().shape, (3, 26, 28))

    # Untied biases require images of a specific size
    x_val_1 = numpy.ones((batch_size, num_channels, 28, 30),
                         dtype=theano.config.floatX)

    assert_allclose(func(x_val_1),
                    numpy.prod(filter_size) * num_channels *
                    numpy.ones((batch_size, num_filters, 26, 28)) + 2)

    x_val_2 = numpy.ones((batch_size, num_channels, 23, 19),
                         dtype=theano.config.floatX)

    def wrongsize():
        func(x_val_2)

    assert_raises_regexp(AssertionError, 'AbstractConv shape mismatch',
                         wrongsize)