diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..4972847 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,4 @@ +[report] +# Ignore abstract methods in base ABC classes +exclude_lines = + raise NotImplementedError\(\"Implemented in subclass!\"\) \ No newline at end of file diff --git a/tests/base.py b/tests/base.py new file mode 100644 index 0000000..821c867 --- /dev/null +++ b/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() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a1bd88c --- /dev/null +++ b/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!") diff --git a/tests/test_dev.py b/tests/test_dev.py new file mode 100644 index 0000000..ce6d7b1 --- /dev/null +++ b/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} diff --git a/tests/test_init.py b/tests/test_init.py deleted file mode 100644 index 7a64bc7..0000000 --- a/tests/test_init.py +++ /dev/null @@ -1,92 +0,0 @@ -from time import sleep -import aiohttp -import signal -import os -import sys -import subprocess -import pytest - - -session: aiohttp.ClientSession - - -@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 - - print("running tests!") - yield # Run tests - - 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!") - - -@pytest.fixture(scope="function", autouse=True) -async def createSession(): - print("Creating session...") - global session - session = aiohttp.ClientSession() - yield - # Close the session if it isn't already closed - # eg. test fails so context is not released - await session.close() - - -@pytest.mark.asyncio -async def test_index(): - async with session.get("http://localhost:3000") as res: - assert res.status == 200 - assert res.content_type == "text/html" - - -@pytest.mark.asyncio -async def test_readme(): - async with 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(): - async with 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] diff --git a/tests/test_root.py b/tests/test_root.py new file mode 100644 index 0000000..aefa0ab --- /dev/null +++ b/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 diff --git a/tests/test_xuid.py b/tests/test_xuid.py new file mode 100644 index 0000000..02ea058 --- /dev/null +++ b/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