def test_wayf():
    c = SPConfig().load_file("server_conf")
    c.context = "sp"

    idps = c.metadata.with_descriptor("idpsso")
    ent = idps.values()[0]
    assert name(ent) == 'Example Co.'
    assert name(ent, "se") == 'Exempel AB'

    c.setup_logger()

    assert root_logger.level != logging.NOTSET
    assert root_logger.level == logging.INFO
    assert len(root_logger.handlers) == 1
    assert isinstance(root_logger.handlers[0],
                        logging.handlers.RotatingFileHandler)
    handler = root_logger.handlers[0]
    assert handler.backupCount == 5
    try:
        assert handler.maxBytes == 100000
    except AssertionError:
        assert handler.maxBytes == 500000
    assert handler.mode == "a"
    assert root_logger.name == "saml2"
    assert root_logger.level == 20

def get_saml2_config(module_path):

    module = imp.load_source('saml2_settings', module_path)

    conf = SPConfig()
    conf.load(module.SAML_CONFIG)
    return conf

def test():
    with closing(Server(config_file=dotname("idp_all_conf"))) as idp:
        conf = SPConfig()
        conf.load_file(dotname("servera_conf"))
        sp = Saml2Client(conf)

        srvs = sp.metadata.single_sign_on_service(idp.config.entityid,
                                                  BINDING_HTTP_REDIRECT)

        destination = srvs[0]["location"]
        req_id, req = sp.create_authn_request(destination, id="id1")

        try:
            key = sp.sec.key
        except AttributeError:
            key = import_rsa_key_from_file(sp.sec.key_file)

        info = http_redirect_message(req, destination, relay_state="RS",
                                     typ="SAMLRequest", sigalg=SIG_RSA_SHA1,
                                     key=key)

        verified_ok = False

        for param, val in info["headers"]:
            if param == "Location":
                _dict = parse_qs(val.split("?")[1])
                _certs = idp.metadata.certs(sp.config.entityid, "any", "signing")
                for cert in _certs:
                    if verify_redirect_signature(_dict, cert):
                        verified_ok = True

        assert verified_ok

def test_conf_syslog():
    c = SPConfig().load_file("server_conf_syslog")
    c.context = "sp"

    # otherwise the logger setting is not changed
    root_logger.level = logging.NOTSET
    root_logger.handlers = []
    
    print c.logger
    c.setup_logger()

    assert root_logger.level != logging.NOTSET
    assert root_logger.level == logging.INFO
    assert len(root_logger.handlers) == 1
    assert isinstance(root_logger.handlers[0],
                        logging.handlers.SysLogHandler)
    handler = root_logger.handlers[0]
    print handler.__dict__
    assert handler.facility == "local3"
    assert handler.address == ('localhost', 514)
    if sys.version >= (2, 7):
        assert handler.socktype == 2
    else:
        pass
    assert root_logger.name == "saml2"
    assert root_logger.level == 20

def test_conf_syslog():
    c = SPConfig().load_file("server_conf_syslog")
    c.context = "sp"

    # otherwise the logger setting is not changed
    root_logger.level = logging.NOTSET
    while root_logger.handlers:
        handler = root_logger.handlers[-1]
        root_logger.removeHandler(handler)
        handler.flush()
        handler.close()

    print(c.logger)
    c.setup_logger()

    assert root_logger.level != logging.NOTSET
    assert root_logger.level == logging.INFO
    assert len(root_logger.handlers) == 1
    assert isinstance(root_logger.handlers[0],
                      logging.handlers.SysLogHandler)
    handler = root_logger.handlers[0]
    print(handler.__dict__)
    assert handler.facility == "local3"
    assert handler.address == ('localhost', 514)
    if ((sys.version_info.major == 2 and sys.version_info.minor >= 7) or
        sys.version_info.major > 2):
        assert handler.socktype == 2
    else:
        pass
    assert root_logger.name == "saml2"
    assert root_logger.level == 20

    def setup_class(self):
        self.server = FakeIDP("idp_all_conf")

        conf = SPConfig()
        conf.load_file("servera_conf")
        self.client = Saml2Client(conf)

        self.client.send = self.server.receive

def config_settings_loader(request=None):
    """Utility function to load the pysaml2 configuration.

    This is also the default config loader.
    """
    conf = SPConfig()
    conf.load(copy.deepcopy(settings.SAML_CONFIG))
    return conf

def make_entity(sp, **kw_args):
    try:
        conf = SPConfig().load(kw_args["spconf"][sp])
    except KeyError:
        logging.warning("known SP configs: {}".format(kw_args["spconf"].keys()))
        raise

    conf.metadata = kw_args['metadata']

    return Saml2Client(config=conf)

def create_logout_request(subject_id, destination, issuer_entity_id,
        req_entity_id, sign=True):
    config = SPConfig()
    config.load(sp_config)
    sp_client = Saml2Client(config=config)
    # construct a request
    logout_request = samlp.LogoutRequest(
        id='a123456',
        version=VERSION,
        destination=destination,
        issuer=saml.Issuer(text=req_entity_id,
            format=saml.NAMEID_FORMAT_ENTITY),
        name_id=saml.NameID(text=subject_id))
    return logout_request

def test_2():
    c = SPConfig().load(sp2)
    c.context = "sp"

    print c
    assert c.endpoints
    assert c.idp
    assert c.optional_attributes
    assert c.name
    assert c.required_attributes

    assert len(c.idp) == 1
    assert c.idp.keys() == [""]
    assert c.idp.values() == ["https://example.com/saml2/idp/SSOService.php"]
    assert c.only_use_keys_in_metadata is None

def test_2():
    c = SPConfig().load(sp2)
    c.context = "sp"

    print(c)
    assert c._sp_endpoints
    assert c.getattr("endpoints", "sp")
    assert c._sp_idp
    assert c._sp_optional_attributes
    assert c.name
    assert c._sp_required_attributes

    assert len(c._sp_idp) == 1
    assert list(c._sp_idp.keys()) == [""]
    assert list(c._sp_idp.values()) == ["https://example.com/saml2/idp/SSOService.php"]
    assert c.only_use_keys_in_metadata is True

 def _saml2_config(self):
     if self._v_config is None:
         sp_config = self._saml2_config_template()
         sp_config['metadata']['local'] = [self.saml2_idp_configfile]
         sp_config['entityid'] = self.saml2_sp_entityid
         sp_config['service']['sp']['name'] = self.saml2_sp_entityid
         sp_config['service']['sp']['url'] = self.saml2_sp_url
         sp_config['service']['sp']['endpoints']['assertion_consumer_service'] = [self.saml2_sp_url,]
         sp_config['service']['sp']['endpoints']['single_logout_service'] = ['%s/logout' % self.saml2_sp_url, BINDING_HTTP_REDIRECT]
         sp_config['service']['sp']['url'] = self.saml2_sp_url
         sp_config['xmlsec_binary'] = self.saml2_xmlsec
         config = SPConfig()
         conf=sp_config.copy()
         config.load(conf)
         self._v_config = config
     return self._v_config

def test_1():
    c = SPConfig().load(sp1)
    c.context = "sp"
    print c
    assert c._sp_endpoints
    assert c._sp_name
    assert c._sp_idp
    md = c.metadata
    assert isinstance(md, MetaData)

    assert len(c._sp_idp) == 1
    assert c._sp_idp.keys() == ["urn:mace:example.com:saml:roland:idp"]
    assert c._sp_idp.values() == [{'single_sign_on_service':
        {'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect':
         'http://localhost:8088/sso/'}}]

    assert c.only_use_keys_in_metadata

def test_minimum():
    minimum = {
        "entityid": "urn:mace:example.com:saml:roland:sp",
        "service": {
            "sp": {
                "endpoints": {"assertion_consumer_service": ["http://sp.example.org/"]},
                "name": "test",
                "idp": {"": "https://example.com/idp/SSOService.php"},
            }
        },
        # "xmlsec_binary" : "/usr/local/bin/xmlsec1",
    }

    c = SPConfig().load(minimum)
    c.context = "sp"

    assert c is not None

    def get_auth_response(self, samlfrontend, context, internal_response, sp_conf, idp_metadata_str):
        sp_config = SPConfig().load(sp_conf, metadata_construction=False)
        resp_args = {
            "name_id_policy": NameIDPolicy(format=NAMEID_FORMAT_TRANSIENT),
            "in_response_to": None,
            "destination": sp_config.endpoint("assertion_consumer_service", binding=BINDING_HTTP_REDIRECT)[0],
            "sp_entity_id": sp_conf["entityid"],
            "binding": BINDING_HTTP_REDIRECT
        }
        request_state = samlfrontend._create_state_data(context, resp_args, "")
        context.state[samlfrontend.name] = request_state

        resp = samlfrontend.handle_authn_response(context, internal_response)

        sp_conf["metadata"]["inline"].append(idp_metadata_str)
        fakesp = FakeSP(sp_config)
        resp_dict = parse_qs(urlparse(resp.message).query)
        return fakesp.parse_authn_request_response(resp_dict["SAMLResponse"][0], BINDING_HTTP_REDIRECT)

 def _saml2_config(self):
     if self._v_config is None:
         sp_config = self._saml2_config_template()
         sp_config["metadata"]["local"] = [self.saml2_idp_configfile]
         sp_config["entityid"] = self.saml2_sp_entityid
         sp_config["service"]["sp"]["name"] = self.saml2_sp_entityid
         sp_config["service"]["sp"]["url"] = self.saml2_sp_url
         sp_config["service"]["sp"]["endpoints"]["assertion_consumer_service"] = [self.saml2_sp_url]
         sp_config["service"]["sp"]["endpoints"]["single_logout_service"] = [
             "%s/logout" % self.saml2_sp_url,
             BINDING_HTTP_REDIRECT,
         ]
         sp_config["service"]["sp"]["url"] = self.saml2_sp_url
         sp_config["xmlsec_binary"] = self.saml2_xmlsec
         config = SPConfig()
         conf = sp_config.copy()
         config.load(conf)
         self._v_config = config
     return self._v_config

    def __init__(self, outgoing, internal_attributes, config, base_url, name):
        """
        :type outgoing:
        (satosa.context.Context, satosa.internal.InternalData) -> satosa.response.Response
        :type internal_attributes: dict[str, dict[str, list[str] | str]]
        :type config: dict[str, Any]
        :type base_url: str
        :type name: str

        :param outgoing: Callback should be called by the module after
                                   the authorization in the backend is done.
        :param internal_attributes: Internal attribute map
        :param config: The module config
        :param base_url: base url of the service
        :param name: name of the plugin
        """
        super().__init__(outgoing, internal_attributes, base_url, name)
        self.config = self.init_config(config)

        sp_config = SPConfig().load(copy.deepcopy(config[self.KEY_SP_CONFIG]), False)
        self.sp = Base(sp_config)

        self.discosrv = config.get(self.KEY_DISCO_SRV)
        self.encryption_keys = []
        self.outstanding_queries = {}
        self.idp_blacklist_file = config.get('idp_blacklist_file', None)

        sp_keypairs = sp_config.getattr('encryption_keypairs', '')
        sp_key_file = sp_config.getattr('key_file', '')
        if sp_keypairs:
            key_file_paths = [pair['key_file'] for pair in sp_keypairs]
        elif sp_key_file:
            key_file_paths = [sp_key_file]
        else:
            key_file_paths = []

        for p in key_file_paths:
            with open(p) as key_file:
                self.encryption_keys.append(key_file.read())

def test_ecp():
    cnf = SPConfig()
    cnf.load(ECP_SP)
    assert cnf.endpoint("assertion_consumer_service") == ["http://lingon.catalogix.se:8087/"]
    eid = cnf.ecp_endpoint("130.239.16.3")
    assert eid == "http://example.com/idp"
    eid = cnf.ecp_endpoint("130.238.20.20")
    assert eid is None

def test_sp():
    cnf = SPConfig()
    cnf.load_file("sp_1_conf")
    assert cnf.single_logout_services("urn:mace:example.com:saml:roland:idp",
                            BINDING_HTTP_POST) == ["http://localhost:8088/slo"]
    assert cnf.endpoint("assertion_consumer_service") == \
                                            ["http://lingon.catalogix.se:8087/"]
    assert len(cnf.idps()) == 1

    def __init__(self, config):
        """Initialize SAML Service Provider.

        Args:
            config (dict): Service Provider config info in dict form
        """
        if config.get('metadata') is not None:
            config['metadata'] = _parse_metadata_dict_to_inline(
                config['metadata'])
        self._config = SPConfig().load(config)
        self._config.setattr('', 'allow_unknown_attributes', True)
        # Set discovery end point, if configured for.
        if config['service']['sp'].get('ds'):
            self.discovery_service_end_point = \
                config['service']['sp'].get('ds')[0]