1import pytest
2
3from webdriver.transport import HTTPWireProtocol
4
5
6def product(a, b):
7    return [(a, item) for item in b]
8
9
10def flatten(l):
11    return [item for x in l for item in x]
12
13
14@pytest.fixture(name="add_browser_capabilities")
15def fixture_add_browser_capabilities(configuration):
16
17    def add_browser_capabilities(capabilities):
18        # Make sure there aren't keys in common.
19        assert not set(configuration["capabilities"]).intersection(set(capabilities))
20        result = dict(configuration["capabilities"])
21        result.update(capabilities)
22
23        return result
24
25    return add_browser_capabilities
26
27
28@pytest.fixture(name="configuration")
29def fixture_configuration(configuration):
30  """Remove "acceptInsecureCerts" from capabilities if it exists.
31
32  Some browser configurations add acceptInsecureCerts capability by default.
33  Remove it during new_session tests to avoid interference.
34  """
35
36  if "acceptInsecureCerts" in configuration["capabilities"]:
37    configuration = dict(configuration)
38    del configuration["capabilities"]["acceptInsecureCerts"]
39  return configuration
40
41@pytest.fixture(name="new_session")
42def fixture_new_session(request, configuration, current_session):
43    """Start a new session for tests which themselves test creating new sessions.
44
45    :param body: The content of the body for the new session POST request.
46
47    :param delete_existing_session: Allows the fixture to delete an already
48     created custom session before the new session is getting created. This
49     is useful for tests which call this fixture multiple times within the
50     same test.
51    """
52    custom_session = {}
53
54    transport = HTTPWireProtocol(
55        configuration["host"], configuration["port"], url_prefix="/",
56    )
57
58    def _delete_session(session_id):
59        transport.send("DELETE", "session/{}".format(session_id))
60
61    def new_session(body, delete_existing_session=False):
62        # If there is an active session from the global session fixture,
63        # delete that one first
64        if current_session is not None:
65            current_session.end()
66
67        if delete_existing_session:
68            _delete_session(custom_session["session"]["sessionId"])
69
70        response = transport.send("POST", "session", body)
71        if response.status == 200:
72            custom_session["session"] = response.body["value"]
73        return response, custom_session.get("session", None)
74
75    yield new_session
76
77    if custom_session.get("session") is not None:
78        _delete_session(custom_session["session"]["sessionId"])
79        custom_session = None
80