Make tests class-based, add more tests post-rewrite

.coveragerc

@@ -0,0 +1,4 @@
+[report]
+# Ignore abstract methods in base ABC classes
+exclude_lines =
+    raise NotImplementedError\(\"Implemented in subclass!\"\)

tests/base.py

@@ -0,0 +1,17 @@
+from aiohttp import ClientSession
+import pytest
+
+
+class BaseTest:
+    # Define class variable for making http requests
+    session: ClientSession
+
+    # Fixture to get a fresh client session for every test run
+    @pytest.fixture(scope="function", autouse=True)
+    async def create_session(self):
+        # workaround so session is accessible to tests, see: https://github.com/pytest-dev/pytest/issues/3869#issuecomment-488276259
+        type(self).session = ClientSession()
+        yield
+        # Close the session if it isn't already closed
+        # eg. test fails so context is not released
+        await self.session.close()

tests/conftest.py

@@ -0,0 +1,62 @@
+from time import sleep, time
+import signal
+import os
+import sys
+import subprocess
+import pytest
+
+# Fixture to spin up the webserver
+@pytest.fixture(scope="session", autouse=True)
+def setup():
+    # Have this process sigint
+    signal.signal(signal.SIGINT, signal.SIG_IGN)
+
+    cwd = os.getcwd()
+    exec = sys.executable
+
+    should_run_coverage = os.environ.get("XBLAPI_RUN_COVERAGE") is "1"
+
+    if should_run_coverage:
+        args = [exec, "-m", "coverage", "run", "--timid", "server.py"]
+    else:
+        args = [exec, "server.py"]
+
+    print("Starting '%s' in dir '%s'..." % (str(args), cwd))
+
+    p = subprocess.Popen(args, stdout=subprocess.PIPE, shell=False, cwd=cwd)
+    print("Waiting 30 seconds for server start")
+
+    if should_run_coverage:
+        print("Measuring code coverage, DO NOT connect debugger!")
+    else:
+        print("You may connect your debugger now.")
+
+    sleep(30)  # Wait for the server to start
+
+    # Make a note of the test start time
+    epoch_start = int(time())
+
+    print("running tests!")
+    yield  # Run tests
+
+    print(
+        "tests completed! Waiting until server has been up for 5 mins to ensure that scheduled jobs run..."
+    )
+
+    # Wait until 5 minutes has elapsed to ensure that scheduled jobs run
+    while int(time() - epoch_start < 300):
+        continue
+        # Continue to wait...
+
+    # 5 minutes has passed! Let's continue...
+
+    print("Tests finished! Sending SIGINT/CTRL+C")
+
+    if os.name == "nt":
+        # Windows does not support sigint
+        p.send_signal(signal.CTRL_C_EVENT)
+    else:
+        p.send_signal(signal.SIGINT)
+
+    p.wait()  # Wait for subprocess to gracefully exit
+    print("DONE!")

tests/test_dev.py

@@ -0,0 +1,13 @@
+import pytest
+
+from base import BaseTest
+
+# Not including /dev/reauth as it is due to be removed soon
+class TestDev(BaseTest):
+    @pytest.mark.asyncio
+    async def test_isauth(self):
+        async with self.session.get("http://localhost:3000/dev/isauth") as res:
+            assert res.status == 200
+            assert res.content_type == "application/json"
+            # We can use the keyword True here as aiohttp will automatically convert from a JSON boolean when parsing
+            assert (await res.json()) == {"authenticated": True}

tests/test_root.py

@@ -0,0 +1,80 @@
+import subprocess
+import pytest
+
+from base import BaseTest
+
+
+class TestRoot(BaseTest):
+    @pytest.mark.asyncio
+    async def test_index(self):
+        async with self.session.get("http://localhost:3000") as res:
+            assert res.status == 200
+            assert res.content_type == "text/html"
+
+    @pytest.mark.asyncio
+    async def test_readme(self):
+        async with self.session.get("http://localhost:3000/readme") as res:
+            assert res.status == 200
+            assert res.content_type == "text/markdown"
+            assert (
+                str(await res.text()).replace("\r", "")
+                == open("README.md", mode="r").read()
+            )
+
+    @pytest.mark.asyncio
+    async def test_info(self):
+        async with self.session.get("http://localhost:3000/info") as res:
+            assert res.status == 200
+            assert res.content_type == "application/json"
+            # Code copied from server.py when env var not available
+            assert (await res.json())["sha"] == str(
+                subprocess.check_output(["git", "rev-parse", "--short", "HEAD"]).strip()
+            ).split("'")[1::2][0]
+
+    @pytest.mark.asyncio
+    async def test_titleinfo(self):
+        # Known attributes for Forza Horizon 2
+        knownAttributes = {
+            "titleId": "446059611",
+            "pfn": "Microsoft.ForzaHorizon2_8wekyb3d8bbwe",
+            "scid": "788f0100-a4bc-46c3-b19b-d1001a96545b",
+        }
+        async with self.session.get(
+            "http://localhost:3000/titleinfo/%s" % knownAttributes["titleId"]
+        ) as res:
+            assert res.status == 200
+            data: dict = await res.json()
+            # TODO: possible for old XDK (launch Xbox One games) to have classic title / product ids?
+            # TODO: Cross-reference data with catalog lookup
+            assert data["titles"][0]["titleId"] == knownAttributes["titleId"]
+            assert data["titles"][0]["modernTitleId"] == knownAttributes["titleId"]
+            assert data["titles"][0]["pfn"] == knownAttributes["pfn"]
+            assert data["titles"][0]["serviceConfigId"] == knownAttributes["scid"]
+
+    @pytest.mark.asyncio
+    async def test_legacysearch(self):
+        async with self.session.get("http://localhost:3000/legacysearch/value") as res:
+            assert res.status == 410
+            assert (await res.json()) == {
+                "error": "legacysearch not currently available",
+                "code": 410,
+            }
+
+    @pytest.mark.asyncio
+    async def test_gamertagcheck_taken(self):
+        async with self.session.get(
+            "http://localhost:3000/gamertag/check/Major Nelson"
+        ) as res:
+            assert res.status == 200
+            assert (await res.json()) == {"available": "false"}
+
+    @pytest.mark.asyncio
+    async def test_gamertagcheck_available(self):
+        async with self.session.get(
+            "http://localhost:3000/gamertag/check/placeholdergt12"
+        ) as res:
+            assert res.status == 200
+            assert (await res.json()) == {"available": "true"}
+
+    # TODO: Unknown test (currently returns 500?)
+    # Likely because of invalid gamertag

tests/test_xuid.py

@@ -0,0 +1,76 @@
+import pytest
+
+from base import BaseTest
+
+
+class TestXuid(BaseTest):
+    knownData = {"gamertag": "Major Nelson", "xuid": "2584878536129841"}
+
+    firstRequestQueriedAt: str
+    firstNonexistentRequestQueriedAt: str
+
+    @pytest.mark.asyncio
+    async def test_gamertag(self):
+        async with self.session.get("http://localhost:3000/xuid/Major Nelson") as res:
+            assert res.status == 200
+            assert res.content_type == "application/json"
+            assert (await res.json()) == self.knownData
+            type(self).firstRequestQueriedAt = res.headers["X-Queried-At"]
+
+    @pytest.mark.asyncio
+    async def test_gamertag_repeat(self):
+        async with self.session.get("http://localhost:3000/xuid/Major Nelson") as res:
+            assert res.status == 200
+            assert res.content_type == "application/json"
+            assert (await res.json()) == self.knownData
+            assert self.firstRequestQueriedAt == res.headers["X-Queried-At"]
+
+    @pytest.mark.asyncio
+    async def test_gamertag_raw(self):
+        async with self.session.get(
+            "http://localhost:3000/xuid/Major Nelson/raw"
+        ) as res:
+            assert res.status == 200
+            assert res.content_type == "text/html"
+            assert (await res.text()) == self.knownData["xuid"]
+            # Can't check X-Queried-At as it is not carried through
+
+    @pytest.mark.asyncio
+    async def test_gamertag_nonexistent(self):
+        async with self.session.get(
+            "http://localhost:3000/xuid/placeholdergt12"
+        ) as res:
+            assert res.status == 404
+            assert res.content_type == "application/json"
+            assert (await res.json()) == {
+                "error": "could not resolve gamertag",
+                "code": 404,
+            }
+            type(self).firstNonexistentRequestQueriedAt = res.headers["X-Queried-At"]
+
+    @pytest.mark.asyncio
+    async def test_gamertag_nonexistent_repeat(self):
+        async with self.session.get(
+            "http://localhost:3000/xuid/placeholdergt12"
+        ) as res:
+            assert res.status == 404
+            assert res.content_type == "application/json"
+            assert (await res.json()) == {
+                "error": "could not resolve gamertag",
+                "code": 404,
+            }
+            assert self.firstNonexistentRequestQueriedAt == res.headers["X-Queried-At"]
+
+    @pytest.mark.asyncio
+    async def test_gamertag_nonexistent_raw(self):
+        async with self.session.get(
+            "http://localhost:3000/xuid/placeholdergt12/raw"
+        ) as res:
+            # On error, raw passes through the response from the usual endpoint.
+            assert res.status == 404
+            assert res.content_type == "application/json"
+            assert (await res.json()) == {
+                "error": "could not resolve gamertag",
+                "code": 404,
+            }
+            # Can't check X-Queried-At as it is not carried through