Merge pull request #30 from simonsobs/dev

Add tests

Author: TaiSakuma

src/nextline_alert/graphql/__init__.py

@@ -0,0 +1,20 @@
+from os import PathLike
+from pathlib import Path
+
+from graphql import parse, print_ast
+
+
+def read_gql(path: PathLike | str) -> str:
+    '''Load a GraphQL query from a file while checking its syntax.'''
+
+    text = Path(path).read_text()
+    parsed = parse(text)
+    reformatted = print_ast(parsed)
+    return reformatted
+
+
+pwd = Path(__file__).resolve().parent
+
+
+sub = pwd / 'queries'
+QUERY_VERSION = read_gql(sub / 'Version.gql')

src/nextline_alert/graphql/queries/Version.gql

@@ -0,0 +1,5 @@
+query Version {
+  alert {
+    version
+  }
+}

tests/schema/__init__.py

@@ -0,0 +1,14 @@
+import pytest
+from strawberry import Schema
+
+from nextline_alert.schema import Query
+
+
+@pytest.fixture(scope='session')
+def schema() -> Schema:
+    '''GraphQL schema
+
+    The scope is `session` because Hypothesis doesn't allow function-scoped
+    fixtures. This is fine as the schema is stateless.
+    '''
+    return Schema(query=Query)

tests/schema/conftest.py

@@ -0,0 +1,12 @@
+from strawberry.types import ExecutionResult
+
+import nextline_alert
+from nextline_alert.graphql import QUERY_VERSION
+from tests.schema.conftest import Schema
+
+
+async def test_schema(schema: Schema) -> None:
+    resp = await schema.execute(QUERY_VERSION)
+    assert isinstance(resp, ExecutionResult)
+    assert resp.data
+    assert resp.data['alert']['version'] == nextline_alert.__version__

tests/schema/queries/__init__.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import logging
 import time
 
 import pytest
@@ -9,16 +10,16 @@
 from nextline_alert.emitter import Emitter
 
 
-def func_success():
+def func_success():  # pragma: no cover
     time.sleep(0.001)
 
 
-def func_raise_ignore():
+def func_raise_ignore():  # pragma: no cover
     time.sleep(0.001)
     raise KeyboardInterrupt
 
 
-def func_raise():
+def func_raise():  # pragma: no cover
     time.sleep(0.001)
     raise ValueError('test')
 
@@ -72,3 +73,39 @@ async def test_emit_alert() -> None:
     description = data['alerts'][0]['annotations']['description']
     assert platform == labels['platform']
     assert 'ValueError' in description
+
+
+@respx.mock
+async def test_emit_exception(caplog: pytest.LogCaptureFixture) -> None:
+    nextline = Nextline(func_raise)
+
+    url = 'http://localhost:5000/alerts'
+    platform = 'pytest'
+    emitter = Emitter(url=url, platform=platform)
+    assert nextline.register(emitter)
+
+    # Mock the HTTP POST request
+    route = respx.post(url).respond(status_code=500)
+
+    # Run a script that raises an exception and fails to emit an alert
+    with caplog.at_level(logging.ERROR):
+        async with nextline:
+            event = asyncio.Event()
+            await nextline.run_continue_and_wait(event)
+
+    # Assert the log message
+    records = [r for r in caplog.records if r.name == Emitter.__module__]
+    assert len(records) == 1
+    assert records[0].levelname == 'ERROR'
+    assert 'Failed to emit alert' in records[0].message
+
+    # Assert the HTTP POST request
+    assert route.called
+    assert len(route.calls) == 1
+    call = route.calls[0]
+    assert url == call.request.url
+    data = json.loads(call.request.content)
+    labels = data['alerts'][0]['labels']
+    description = data['alerts'][0]['annotations']['description']
+    assert platform == labels['platform']
+    assert 'ValueError' in description

tests/schema/queries/test_version.py

@@ -0,0 +1,21 @@
+import asyncio
+from collections.abc import AsyncIterator
+
+import pytest
+
+from nextline_alert.graphql import QUERY_VERSION
+from nextlinegraphql import create_app
+from nextlinegraphql.plugins.graphql.test import TestClient, gql_request
+
+
+@pytest.fixture
+async def client() -> AsyncIterator[TestClient]:
+    app = create_app()  # the plugin is loaded here
+    async with TestClient(app) as y:
+        await asyncio.sleep(0)
+        yield y
+
+
+async def test_plugin(client: TestClient) -> None:
+    data = await gql_request(client, QUERY_VERSION)
+    assert data