diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_offline.py b/tests/test_offline.py new file mode 100644 index 0000000..f0ad4ac --- /dev/null +++ b/tests/test_offline.py @@ -0,0 +1,119 @@ +"""Offline unit tests: no network, no cloud or LLM credentials, no provider SDKs. They cover the pieces that read +configuration and on-disk data and the parser for LLM replies.""" +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +import aestimare # noqa: E402 +from analyzers import llm_analyzer # noqa: E402 + +EXAMPLE = ROOT / "config" / "accounts.example.yaml" + + +class LoadConfigTest(unittest.TestCase): + def test_example_config_has_the_sections_the_tool_reads(self): + config = aestimare.load_config(str(EXAMPLE)) + self.assertIn("assessment", config) + self.assertTrue(config["aws_accounts"]) + for account in config["aws_accounts"]: + for key in ("name", "profile", "account_id", "regions", "enabled"): + self.assertIn(key, account) + self.assertRegex(str(account["account_id"]), r"^\d{12}$") + self.assertTrue(config["github"]["organizations"]) + + def test_missing_config_falls_back_to_the_example_next_to_it(self): + with tempfile.TemporaryDirectory() as tmp: + config_dir = Path(tmp) + (config_dir / "accounts.example.yaml").write_text("assessment:\n name: fallback\n", encoding="utf-8") + with mock.patch("builtins.print"): + config = aestimare.load_config(str(config_dir / "accounts.yaml")) + self.assertEqual(config["assessment"]["name"], "fallback") + + def test_missing_config_without_an_example_is_an_error(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(FileNotFoundError): + aestimare.load_config(str(Path(tmp) / "accounts.yaml")) + + +class DataFilesTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.config = {"data_dir": self.tmp.name} + + def test_saved_analysis_is_loaded_back_under_its_key(self): + aestimare.save_analysis("security_analysis.json", {"findings": [1, 2]}, self.config) + aestimare.save_analysis("cost_analysis.json", {"when": __import__("datetime").datetime(2026, 1, 2)}, self.config) + loaded = aestimare.load_existing_analysis(self.config) + self.assertEqual(loaded["security"], {"findings": [1, 2]}) + self.assertEqual(loaded["cost"], {"when": "2026-01-02 00:00:00"}) + self.assertNotIn("infrastructure", loaded) + + def test_no_data_directory_loads_as_empty(self): + missing = {"data_dir": str(Path(self.tmp.name) / "absent")} + self.assertEqual(aestimare.load_existing_analysis(missing), {}) + self.assertEqual(aestimare.load_existing_data(missing), ({}, {})) + + def test_collected_data_is_keyed_by_account_and_organisation(self): + base = Path(self.tmp.name) + for kind, name, payload in (("aws", "prod", {"a": 1}), ("github", "org", {"b": 2})): + folder = base / kind / name + folder.mkdir(parents=True) + (folder / "full_data.json").write_text(json.dumps(payload), encoding="utf-8") + (base / "aws" / "empty").mkdir() + (base / "aws" / "stray.txt").write_text("x", encoding="utf-8") + self.assertEqual(aestimare.load_existing_data(self.config), ({"prod": {"a": 1}}, {"org": {"b": 2}})) + + +class ParseJsonResponseTest(unittest.TestCase): + def setUp(self): + with mock.patch.dict(os.environ, {"ANTHROPIC_API_KEY": "unused-test-key"}): + self.analyzer = llm_analyzer.LLMAnalyzer({"llm": {"provider": "anthropic"}}) + + def test_fenced_json_block_is_extracted_from_prose(self): + self.assertEqual(self.analyzer._parse_json_response('Here:\n```json\n{"score": 7}\n```\nDone'), {"score": 7}) + + def test_bare_json_is_parsed(self): + self.assertEqual(self.analyzer._parse_json_response('{"score": 3}'), {"score": 3}) + + def test_a_broken_fence_falls_back_to_the_whole_reply_then_to_raw(self): + broken = '```json\n{"score": }\n```' + with self.assertLogs(llm_analyzer.logger, level="WARNING"): + self.assertEqual(self.analyzer._parse_json_response(broken), {"raw_response": broken}) + + def test_unparseable_text_is_kept_raw_and_logged(self): + with self.assertLogs(llm_analyzer.logger, level="WARNING"): + self.assertEqual(self.analyzer._parse_json_response("no json"), {"raw_response": "no json"}) + + +class ProviderSelectionTest(unittest.TestCase): + def test_unknown_provider_is_rejected(self): + with self.assertRaisesRegex(ValueError, "Unknown LLM provider"): + llm_analyzer.LLMAnalyzer({"llm": {"provider": "nope"}}) + + def test_anthropic_without_a_key_is_rejected(self): + with mock.patch.dict(os.environ, {}, clear=True): + with self.assertRaisesRegex(ValueError, "ANTHROPIC_API_KEY"): + llm_analyzer.LLMAnalyzer({"llm": {"provider": "anthropic"}}) + + def test_local_provider_needs_no_credentials_and_takes_its_endpoint_from_config(self): + analyzer = llm_analyzer.LLMAnalyzer({"llm": {"provider": "local", "base_url": "http://h:1", "model": "m"}}) + self.assertEqual((analyzer.provider.base_url, analyzer.provider.model), ("http://h:1", "m")) + + def test_missing_prompt_template_is_an_error(self): + with mock.patch.dict(os.environ, {"ANTHROPIC_API_KEY": "unused-test-key"}): + analyzer = llm_analyzer.LLMAnalyzer({"llm": {}}) + with self.assertRaises(FileNotFoundError): + analyzer._load_prompt("does-not-exist.md") + self.assertTrue(analyzer._load_prompt("security_assessment.md")) + + +if __name__ == "__main__": + unittest.main()