import json import os from pathlib import Path import tempfile import time import unittest from unittest.mock import patch from metric_format import application_samples, disk_usage, QUEUES, SCOPES, STATES, CLASSES from metrics_core import MetricsJob, SERVICES from transport import ComposeStack, OperationsError def payload(): rows = [{"name": "post76_queue_jobs", "labels": {"queue": queue, "state": state}, "value": 3} for queue in QUEUES for state in STATES] rows += [{"name": "post76_http_responses_5m", "labels": {"scope": scope, "status_class": status}, "value": 2} for scope in SCOPES for status in CLASSES] rows += [{"name": "post76_queue_failed_jobs_5m", "labels": {}, "value": 0}, {"name": "post76_disk_total_bytes", "labels": {"volume": "uploads"}, "value": 100000}, {"name": "post76_disk_free_bytes", "labels": {"volume": "uploads"}, "value": 75000}] return {"format": 1, "collected_at": int(time.time()), "samples": rows} class FakeMetricsStack: project = "synthetic-post76" def __init__(self): self.fail = False self.calls = [] self.payload = payload() def model(self): return {} def state(self): return {service: {"State": "running", "Health": "healthy"} for service in SERVICES} def command(self, arguments, step): self.calls.append(arguments) if self.fail: raise OperationsError("Synthetic private error must not be exported") if "docker/metrics.php" in arguments: return json.dumps(self.payload).encode() return b"Filesystem 1024-blocks Used Available Capacity Mounted on\n/dev/synthetic 1000 250 750 25% /data\n" class MetricsTest(unittest.TestCase): def setUp(self): parent = Path(__file__).resolve().parent.parent / "storage/app/private/monitoring-check" parent.mkdir(parents=True, exist_ok=True) self.temporary = tempfile.TemporaryDirectory(prefix="unit-", dir=parent) self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.stack = FakeMetricsStack() self.job = MetricsJob(self.stack, self.root / "workspace") def test_read_only_collection_exports_aggregate_counts_and_private_backup_status(self): (self.job.workspace / "status.json").write_text(json.dumps({"success": True, "resumed": True, "last_success": 1000, "duration": 3.2, "snapshot": "synthetic-never-export-this"})) self.job.collect() output = (self.job.workspace / "metrics/metrics.prom").read_text() self.assertIn('post76_queue_jobs{project="synthetic-post76",queue="mail",state="ready"} 3', output) self.assertIn('post76_backup_last_success_timestamp_seconds{project="synthetic-post76"} 1000', output) self.assertNotIn("synthetic-never-export-this", output) self.assertNotIn("/dev/synthetic", output) self.assertTrue(all(arguments[0] == "exec" for arguments in self.stack.calls)) self.assertEqual(0o644, (self.job.workspace / "metrics/metrics.prom").stat().st_mode & 0o777) self.assertEqual(0o600, self.job.state.stat().st_mode & 0o777) def test_failure_preserves_last_good_values_and_timestamp_and_marks_them_stale(self): self.job.collect() previous = json.loads(self.job.state.read_text()) self.stack.fail = True with self.assertRaises(OperationsError): self.job.collect() self.assertEqual(previous, json.loads(self.job.state.read_text())) output = (self.job.workspace / "metrics/metrics.prom").read_text() self.assertIn('post76_collector_success{project="synthetic-post76"} 0', output) self.assertIn('queue="mail",state="ready"} 3', output) self.assertNotIn("Synthetic private error", output) def test_initial_failure_never_fabricates_zero_queue_or_disk_samples(self): self.stack.fail = True with self.assertRaises(OperationsError): self.job.collect() output = (self.job.workspace / "metrics/metrics.prom").read_text() self.assertNotIn("post76_queue_jobs", output) self.assertNotIn("post76_disk_free_bytes", output) self.assertIn('post76_collector_last_success_timestamp_seconds{project="synthetic-post76"} 0', output) def test_maintenance_remains_visible_during_application_collection_failure(self): (self.job.workspace / "maintenance.json").write_text("{}") self.stack.fail = True with self.assertRaises(OperationsError): self.job.collect() self.assertIn('post76_backup_maintenance_active{project="synthetic-post76"} 1', (self.job.workspace / "metrics/metrics.prom").read_text()) def test_private_labels_duplicates_missing_metrics_and_nan_are_rejected(self): for mutate in [lambda rows: rows[0]["labels"].update({"member": "synthetic-private-email"}), lambda rows: rows.append(rows[0]), lambda rows: rows.pop(), lambda rows: rows[0].update({"value": float("nan")})]: values = payload() mutate(values["samples"]) with self.assertRaises(OperationsError): application_samples(values) def test_future_clock_and_malformed_disk_output_fail_collection(self): self.stack.payload["collected_at"] += 1000 with self.assertRaises(OperationsError): self.job.collect() self.assertFalse(self.job.state.exists()) with self.assertRaises(OperationsError): disk_usage(b"malformed disk response") def test_collector_transport_needs_no_backup_repository_and_cannot_inherit_credentials(self): source = self.root / "synthetic.env" source.write_text("APP_ENV=testing\n") with patch.dict(os.environ, {"PATH": "/usr/bin", "DOCKER_HOST": "tcp://forbidden.invalid", "AWS_ACCESS_KEY_ID": "synthetic"}, clear=True): stack = ComposeStack(self.root, "synthetic-post76", source, self.job.workspace, operations=False) self.assertEqual("unix:///var/run/docker.sock", stack.environment["DOCKER_HOST"]) self.assertNotIn("AWS_ACCESS_KEY_ID", stack.environment) self.assertFalse(any(path.name == "compose.backup.yaml" for path in stack.files)) with patch.object(stack, "command", return_value=b'{"services": {}}'): self.assertEqual({"services": {}}, stack.model()) def test_overlapping_collectors_are_refused(self): with self.job.lock(): with self.assertRaises(OperationsError): with self.job.lock(): self.fail("Overlapping collector was accepted") def test_service_failures_remain_fresh_when_application_metrics_fail(self): self.job.collect() self.stack.fail = True states = self.stack.state() states["mysql"]["State"] = "exited" with patch.object(self.stack, "state", return_value=states): with self.assertRaises(OperationsError): self.job.collect() output = (self.job.workspace / "metrics/metrics.prom").read_text() self.assertIn('post76_service_running{project="synthetic-post76",service="mysql"} 0', output) self.assertIn('post76_service_status_success{project="synthetic-post76"} 1', output) def test_remote_prometheus_configuration_requires_https_and_excludes_credentials(self): from render_monitoring import configuration values = configuration("https://metrics.example.invalid:443/metrics") self.assertEqual(["metrics.example.invalid:443"], values["scrape_configs"][0]["static_configs"][0]["targets"]) self.assertFalse(values["scrape_configs"][0]["tls_config"]["insecure_skip_verify"]) for endpoint in ["http://metrics.example.invalid/metrics", "https://user:synthetic@metrics.example.invalid/metrics", "https://metrics.example.invalid/metrics?key=synthetic", "https://metrics.example.invalid/private"]: with self.assertRaises(ValueError): configuration(endpoint) def test_glitchtip_startup_reads_only_explicit_synthetic_files_and_rejects_bad_keys(self): import importlib.util script = Path(__file__).resolve().parent.parent / "docker/monitoring/glitchtip.py" spec = importlib.util.spec_from_file_location("synthetic_glitchtip", script) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) (self.root / "glitchtip_secret_key").write_text("a" * 64) password = self.root / "glitchtip_db_password" password.write_text("b" * 64) values = module.environment({"POST76_MONITORING_SECRET_PATH": str(self.root)}) self.assertEqual("all_in_one", values["SERVER_ROLE"]) self.assertEqual("a" * 64, values["SECRET_KEY"]) self.assertIn("glitchtip-db:5432/glitchtip", values["DATABASE_URL"]) password.write_text("synthetic-invalid-key") with self.assertRaises(ValueError) as result: module.environment({"POST76_MONITORING_SECRET_PATH": str(self.root)}) self.assertNotIn("synthetic-invalid-key", str(result.exception)) if __name__ == "__main__": unittest.main()