diff --git a/tests/conftest.py b/tests/conftest.py index 481eb2a..fdae88e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,48 +1,59 @@ import ssl +from typing import Iterator import pytest -import trustme +from trustme import ( + CA, + LeafCert, +) @pytest.fixture -def ca(): - yield trustme.CA() +def ca() -> Iterator[CA]: + """A root CA.""" + yield CA() @pytest.fixture -def tls_ca_path(ca): +def tls_ca_path(ca: CA) -> Iterator[str]: + """Path for the CA certificate.""" with ca.cert_pem.tempfile() as ca_cert_pem: yield ca_cert_pem @pytest.fixture -def tls_certificate(ca): +def tls_certificate(ca: CA) -> Iterator[LeafCert]: + """A leaf certificate.""" yield ca.issue_cert("localhost", "127.0.0.1", "::1") @pytest.fixture -def tls_public_key_path(tls_certificate): +def tls_public_key_path(tls_certificate: LeafCert) -> Iterator[str]: """Provide a certificate chain PEM file path via fixture.""" with tls_certificate.private_key_and_cert_chain_pem.tempfile() as cert_pem: yield cert_pem @pytest.fixture -def tls_private_key_path(tls_certificate): +def tls_private_key_path(tls_certificate: LeafCert) -> Iterator[str]: """Provide a certificate private key PEM file path via fixture.""" with tls_certificate.private_key_pem.tempfile() as cert_key_pem: yield cert_key_pem @pytest.fixture -def ssl_context(tls_certificate): +def ssl_context(tls_certificate: LeafCert) -> Iterator[ssl.SSLContext]: + """SSL context with the test CA.""" ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) tls_certificate.configure_cert(ssl_ctx) yield ssl_ctx @pytest.fixture -def ssl_context_server(tls_public_key_path, ca): +def ssl_context_server( + ca: CA, tls_public_key_path: str +) -> Iterator[ssl.SSLContext]: + """SSL context for server authentication.""" ssl_ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH) ca.configure_trust(ssl_ctx) yield ssl_ctx diff --git a/tests/metric_test.py b/tests/metric_test.py index ec64ab9..90d7376 100644 --- a/tests/metric_test.py +++ b/tests/metric_test.py @@ -1,8 +1,10 @@ from typing import ( Any, Callable, + cast, ) +from prometheus_client import Histogram from prometheus_client.metrics import MetricWrapperBase import pytest @@ -14,8 +16,7 @@ class TestMetricConfig: - def test_invalid_metric_type(self): - """An invalid metric type raises an error.""" + def test_invalid_metric_type(self) -> None: with pytest.raises(InvalidMetricType) as error: MetricConfig("m1", "desc1", "unknown") assert str(error.value) == ( @@ -29,8 +30,7 @@ def test_labels_sorted(self) -> None: class TestMetricsRegistry: - def test_create_metrics(self): - """Prometheus metrics are created from the specified config.""" + def test_create_metrics(self) -> None: configs = [ MetricConfig("m1", "desc1", "counter"), MetricConfig("m2", "desc2", "histogram"), @@ -40,27 +40,25 @@ def test_create_metrics(self): assert metrics["m1"]._type == "counter" assert metrics["m2"]._type == "histogram" - def test_create_metrics_with_config(self): - """Metric configs are applied.""" + def test_create_metrics_with_config(self) -> None: configs = [ MetricConfig( "m1", "desc1", "histogram", config={"buckets": [10, 20]} ) ] metrics = MetricsRegistry().create_metrics(configs) - # The two specified bucket plus +Inf - assert len(metrics["m1"]._buckets) == 3 + # Histogram has the two specified bucket plus +Inf + histogram = cast(Histogram, metrics["m1"]) + assert len(histogram._buckets) == 3 - def test_create_metrics_config_ignores_unknown(self): - """Unknown metric configs are ignored and don't cause an error.""" + def test_create_metrics_config_ignores_unknown(self) -> None: configs = [ MetricConfig("m1", "desc1", "gauge", config={"unknown": "value"}) ] metrics = MetricsRegistry().create_metrics(configs) assert len(metrics) == 1 - def test_get_metrics(self): - """get_metrics returns a dict with metrics.""" + def test_get_metrics(self) -> None: registry = MetricsRegistry() metrics = registry.create_metrics( [ @@ -70,8 +68,7 @@ def test_get_metrics(self): ) assert registry.get_metrics() == metrics - def test_get_metric(self): - """get_metric returns a metric.""" + def test_get_metric(self) -> None: configs = [ MetricConfig( "m", @@ -86,8 +83,7 @@ def test_get_metric(self): assert metric._name == "m" assert metric._labelvalues == () - def test_get_metric_with_labels(self): - """get_metric returns a metric configured with labels.""" + def test_get_metric_with_labels(self) -> None: configs = [ MetricConfig("m", "A test gauge", "gauge", labels=("l1", "l2")) ] diff --git a/tests/script_test.py b/tests/script_test.py index 3459378..d715aad 100644 --- a/tests/script_test.py +++ b/tests/script_test.py @@ -1,9 +1,12 @@ +from argparse import ArgumentParser from io import StringIO import logging from ssl import SSLContext +from typing import Iterator from unittest import mock import pytest +from pytest_mock import MockerFixture from prometheus_aioexporter._metric import MetricConfig from prometheus_aioexporter._script import PrometheusExporterScript @@ -15,41 +18,36 @@ class SampleScript(PrometheusExporterScript): name = "sample-script" default_port = 12345 + def configure_argument_parser(self, parser: ArgumentParser) -> None: + parser.add_argument("--test", help="test argument") + @pytest.fixture -def script(): +def script() -> Iterator[PrometheusExporterScript]: yield SampleScript() class TestPrometheusExporterScript: - def test_description(self, script): - """The description attribute returns the class docstring.""" + def test_description(self, script: PrometheusExporterScript) -> None: assert script.description == "A sample script" - def test_description_empty(self, script): - """The description is empty string if no docstring is set.""" + def test_description_empty(self, script: PrometheusExporterScript) -> None: script.__doc__ = None assert script.description == "" - def test_logger(self, script): - """The script logger uses the script name.""" + def test_logger(self, script: PrometheusExporterScript) -> None: assert script.logger.name == "sample-script" - def test_configure_argument_parser(self, script): - """configure_argument_parser adds specified arguments.""" - - def configure_argument_parser(parser): - parser.add_argument("test", help="test argument") - - script.configure_argument_parser = configure_argument_parser + def test_configure_argument_parser( + self, script: PrometheusExporterScript + ) -> None: parser = script.get_parser() fh = StringIO() parser.print_help(file=fh) assert "test argument" in fh.getvalue() - def test_create_metrics(self, script): - """Metrics are created based on the configuration.""" + def test_create_metrics(self, script: PrometheusExporterScript) -> None: configs = [ MetricConfig("m1", "desc1", "counter", {}), MetricConfig("m2", "desc2", "histogram", {}), @@ -59,8 +57,9 @@ def test_create_metrics(self, script): assert metrics["m1"]._type == "counter" assert metrics["m2"]._type == "histogram" - def test_setup_logging(self, mocker, script): - """Logging is set up.""" + def test_setup_logging( + self, mocker: MockerFixture, script: PrometheusExporterScript + ) -> None: mock_setup_logger = mocker.patch( "prometheus_aioexporter._script.setup_logger" ) @@ -79,24 +78,27 @@ def test_setup_logging(self, mocker, script): ] mock_setup_logger.assert_has_calls(calls) - def test_change_metrics_path(self, script): - """The path under which metrics are exposed can be changed.""" + def test_change_metrics_path( + self, script: PrometheusExporterScript + ) -> None: args = script.get_parser().parse_args( ["--metrics-path", "/other-path"] ) exporter = script._get_exporter(args) assert exporter.metrics_path == "/other-path" - def test_only_ssl_key(self, script, tls_private_key_path): - """The path under which metrics are exposed can be changed.""" + def test_only_ssl_key( + self, script: PrometheusExporterScript, tls_private_key_path: str + ) -> None: args = script.get_parser().parse_args( ["--ssl-private-key", tls_private_key_path] ) exporter = script._get_exporter(args) assert exporter.ssl_context is None - def test_only_ssl_cert(self, script, tls_public_key_path): - """The path under which metrics are exposed can be changed.""" + def test_only_ssl_cert( + self, script: PrometheusExporterScript, tls_public_key_path: str + ) -> None: args = script.get_parser().parse_args( ["--ssl-public-key", tls_public_key_path] ) @@ -104,9 +106,11 @@ def test_only_ssl_cert(self, script, tls_public_key_path): assert exporter.ssl_context is None def test_ssl_components_without_ca( - self, script, tls_private_key_path, tls_public_key_path - ): - """The path under which metrics are exposed can be changed.""" + self, + script: PrometheusExporterScript, + tls_private_key_path: str, + tls_public_key_path: str, + ) -> None: args = script.get_parser().parse_args( [ "--ssl-public-key", @@ -120,9 +124,12 @@ def test_ssl_components_without_ca( assert len(exporter.ssl_context.get_ca_certs()) != 1 def test_ssl_components( - self, script, tls_private_key_path, tls_ca_path, tls_public_key_path - ): - """The path under which metrics are exposed can be changed.""" + self, + script: PrometheusExporterScript, + tls_private_key_path: str, + tls_ca_path: str, + tls_public_key_path: str, + ) -> None: args = script.get_parser().parse_args( [ "--ssl-public-key", @@ -137,8 +144,9 @@ def test_ssl_components( assert isinstance(exporter.ssl_context, SSLContext) assert len(exporter.ssl_context.get_ca_certs()) == 1 - def test_include_process_stats(self, mocker, script): - """The script can include process stats in metrics.""" + def test_include_process_stats( + self, mocker: MockerFixture, script: PrometheusExporterScript + ) -> None: mocker.patch("prometheus_aioexporter._web.PrometheusExporter.run") script(["--process-stats"]) # process stats are present in the registry @@ -147,8 +155,9 @@ def test_include_process_stats(self, mocker, script): in script.registry.registry._names_to_collectors ) - def test_get_exporter_registers_handlers(self, script): - """Startup/shutdown handlers are registered with the application.""" + def test_get_exporter_registers_handlers( + self, script: PrometheusExporterScript + ) -> None: args = script.get_parser().parse_args([]) exporter = script._get_exporter(args) assert script.on_application_startup in exporter.app.on_startup @@ -156,13 +165,12 @@ def test_get_exporter_registers_handlers(self, script): def test_script_run_exporter_ssl( self, - mocker, - script, - ssl_context, - tls_private_key_path, - tls_public_key_path, - ): - """The script runs the exporter application.""" + mocker: MockerFixture, + script: PrometheusExporterScript, + ssl_context: SSLContext, + tls_private_key_path: str, + tls_public_key_path: str, + ) -> None: mock_run_app = mocker.patch("prometheus_aioexporter._web.run_app") script( [ @@ -177,8 +185,9 @@ def test_script_run_exporter_ssl( mock_run_app.call_args.kwargs["ssl_context"], SSLContext ) - def test_script_run_exporter(self, mocker, script): - """The script runs the exporter application.""" + def test_script_run_exporter( + self, mocker: MockerFixture, script: PrometheusExporterScript + ) -> None: mock_run_app = mocker.patch("prometheus_aioexporter._web.run_app") script([]) mock_run_app.assert_called_with( diff --git a/tests/web_test.py b/tests/web_test.py index 47e75ee..cb2a5ac 100644 --- a/tests/web_test.py +++ b/tests/web_test.py @@ -1,7 +1,23 @@ from ssl import SSLContext +from typing import ( + Any, + Awaitable, + Callable, + cast, + Coroutine, + Iterator, +) from unittest import mock +from aiohttp.test_utils import ( + TestClient, + TestServer, +) +from aiohttp.web import Application +from prometheus_client import Gauge +from prometheus_client.metrics import MetricWrapperBase import pytest +from pytest_mock import MockerFixture from prometheus_aioexporter._metric import ( MetricConfig, @@ -10,56 +26,53 @@ from prometheus_aioexporter._web import PrometheusExporter from tests.conftest import ssl_context - -@pytest.fixture -def registry(): - yield MetricsRegistry() +AiohttpClient = Callable[[Application | TestServer], Awaitable[TestClient]] +AiohttpServer = Callable[..., Awaitable[TestServer]] @pytest.fixture -def exporter(registry, request): - yield PrometheusExporter( - name="test-app", - description="A test application", - hosts=["localhost"], - port=8000, - registry=registry, - ssl_context=request.param, - ) +def registry() -> Iterator[MetricsRegistry]: + yield MetricsRegistry() @pytest.fixture -def exporter_ssl(registry, ssl_context): +def exporter( + registry: MetricsRegistry, request: pytest.FixtureRequest +) -> Iterator[PrometheusExporter]: yield PrometheusExporter( name="test-app", description="A test application", hosts=["localhost"], port=8000, registry=registry, - ssl_context=ssl_context, + ssl_context=getattr(request, "param", None), ) @pytest.fixture -def create_server_client(ssl_context, aiohttp_server): - def create(exporter): - kwargs = {} +def create_server_client( + ssl_context: SSLContext, + aiohttp_server: AiohttpServer, +) -> Iterator[Callable[[PrometheusExporter], Coroutine[Any, Any, TestServer]]]: + async def create(exporter: PrometheusExporter) -> TestServer: + kwargs: dict[str, Any] = {} if exporter.ssl_context is None: kwargs["ssl"] = exporter.ssl_context - return aiohttp_server(exporter.app, **kwargs) + return await aiohttp_server(exporter.app, **kwargs) - return create + yield create class TestPrometheusExporter: - @pytest.mark.parametrize("exporter", [ssl_context, None], indirect=True) - def test_app_exporter_reference(self, exporter): - """The application has a reference to the exporter.""" + def test_app_exporter_reference( + self, exporter: PrometheusExporter + ) -> None: assert exporter.app["exporter"] is exporter @pytest.mark.parametrize("exporter", [ssl_context, None], indirect=True) - def test_run(self, mocker, exporter): - """The script starts the web application.""" + def test_run( + self, mocker: MockerFixture, exporter: PrometheusExporter + ) -> None: mock_run_app = mocker.patch("prometheus_aioexporter._web.run_app") exporter.run() mock_run_app.assert_called_with( @@ -74,12 +87,13 @@ def test_run(self, mocker, exporter): @pytest.mark.parametrize("exporter", [ssl_context, None], indirect=True) async def test_homepage( self, - ssl_context_server, - create_server_client, - exporter, - aiohttp_client, - ): - """The homepage shows an HTML page.""" + ssl_context_server: SSLContext, + create_server_client: Callable[ + [PrometheusExporter], Coroutine[Any, Any, TestServer] + ], + exporter: PrometheusExporter, + aiohttp_client: AiohttpClient, + ) -> None: server = await create_server_client(exporter) client = await aiohttp_client(server) ssl_client_context = None @@ -91,10 +105,12 @@ async def test_homepage( text = await request.text() assert "test-app - A test application" in text - @pytest.mark.parametrize("exporter", [ssl_context, None], indirect=True) - async def test_homepage_no_description(self, aiohttp_client, exporter): - """The title is set to just the name if no description is present.""" - exporter.description = None + async def test_homepage_no_description( + self, + aiohttp_client: AiohttpClient, + exporter: PrometheusExporter, + ) -> None: + exporter.description = "" client = await aiohttp_client(exporter.app) request = await client.request("GET", "/") assert request.status == 200 @@ -102,13 +118,17 @@ async def test_homepage_no_description(self, aiohttp_client, exporter): text = await request.text() assert "test-app" in text - @pytest.mark.parametrize("exporter", [ssl_context, None], indirect=True) - async def test_metrics(self, aiohttp_client, exporter, registry): - """The /metrics page display Prometheus metrics.""" + async def test_metrics( + self, + aiohttp_client: AiohttpClient, + exporter: PrometheusExporter, + registry: MetricsRegistry, + ) -> None: metrics = registry.create_metrics( [MetricConfig("test_gauge", "A test gauge", "gauge")] ) - metrics["test_gauge"].set(12.3) + gauge = cast(Gauge, metrics["test_gauge"]) + gauge.set(12.3) client = await aiohttp_client(exporter.app) request = await client.request("GET", "/metrics") assert request.status == 200 @@ -117,24 +137,26 @@ async def test_metrics(self, aiohttp_client, exporter, registry): assert "HELP test_gauge A test gauge" in text assert "test_gauge 12.3" in text - @pytest.mark.parametrize("ssl_context", [SSLContext(), None]) async def test_metrics_different_path( - self, aiohttp_client, registry, ssl_context - ): - """The metrics path can be changed.""" + self, + aiohttp_client: AiohttpClient, + registry: MetricsRegistry, + ssl_context: SSLContext, + ) -> None: exporter = PrometheusExporter( - "test-app", - "A test application", - "localhost", - 8000, - registry, + name="test-app", + description="A test application", + hosts=["localhost"], + port=8000, + registry=registry, metrics_path="/other-path", ssl_context=ssl_context, ) metrics = registry.create_metrics( [MetricConfig("test_gauge", "A test gauge", "gauge")] ) - metrics["test_gauge"].set(12.3) + gaute = cast(Gauge, metrics["test_gauge"]) + gaute.set(12.3) client = await aiohttp_client(exporter.app) request = await client.request("GET", "/other-path") assert request.status == 200 @@ -146,14 +168,17 @@ async def test_metrics_different_path( request = await client.request("GET", "/metrics") assert request.status == 404 - @pytest.mark.parametrize("exporter", [ssl_context, None], indirect=True) async def test_metrics_update_handler( - self, aiohttp_client, exporter, registry - ): - """set_metric_update_handler sets a handler called with metrics.""" + self, + aiohttp_client: AiohttpClient, + exporter: PrometheusExporter, + registry: MetricsRegistry, + ) -> None: args = [] - async def update_handler(metrics): + async def update_handler( + metrics: dict[str, MetricWrapperBase] + ) -> None: args.append(metrics) exporter.set_metric_update_handler(update_handler) @@ -171,8 +196,12 @@ async def update_handler(metrics): ["ssl_context", "protocol"], [(ssl_context, "https"), (None, "http")] ) async def test_startup_logger( - self, mocker, registry, ssl_context, protocol - ): + self, + mocker: MockerFixture, + registry: MetricsRegistry, + ssl_context: SSLContext, + protocol: str, + ) -> None: exporter = PrometheusExporter( "test-app", "A test application", diff --git a/tox.ini b/tox.ini index 3366565..9671be7 100644 --- a/tox.ini +++ b/tox.ini @@ -14,7 +14,7 @@ deps = .[testing] mypy commands = - mypy prometheus_aioexporter {posargs} + mypy {[base]lint_files} {posargs} [testenv:coverage] deps =