"""Unit tests for the application-database qualifier and ODBC server form.""" from __future__ import annotations import ast import unittest from datetime import date, timedelta from pathlib import Path from unittest.mock import MagicMock, patch from pipeline.config import SqlSettings from pipeline.db import _connection_string, app_table, qualify_app_sql from pipeline.stages.advice import AdviceResult, insert_advice from pipeline.stages.advice_context import load_last_advice, load_previous_incubation_register from pipeline.stages.incubations import incubation_start_covered from pipeline.stages.product_json import ProductJsonBuilder from pipeline.stages.products import ProductIndex, ProductRow, last_treatment_interval_open from pipeline.stages.treatments import TreatmentHistory _ROOT = Path(__file__).resolve().parent.parent def _sql(**overrides: str) -> SqlSettings: values = { "driver": "ODBC Driver 17 for SQL Server", "server": "localhost", "database": "DBMeteo", "app_database": "DBMeteo", "username": "user", "password": "secret", } values.update(overrides) return SqlSettings(**values) class AppTableTests(unittest.TestCase): def test_same_database_stays_bare(self) -> None: self.assertEqual(app_table(_sql(), "advisories"), "advisories") self.assertEqual(app_table(_sql(app_database="dbmeteo"), "products"), "products") def test_different_database_is_three_part(self) -> None: settings = _sql(app_database="Agrosupport") self.assertEqual(app_table(settings, "advisories"), "[Agrosupport].dbo.[advisories]") def test_qualify_leaves_placeholders(self) -> None: sql = "SELECT product_id FROM {product_uses} WHERE product_id IN ({placeholders})" qualified = qualify_app_sql(_sql(app_database="Agrosupport"), sql) self.assertEqual( qualified, "SELECT product_id FROM [Agrosupport].dbo.[product_uses] " "WHERE product_id IN ({placeholders})", ) def test_unknown_table_rejected(self) -> None: with self.assertRaises(ValueError): app_table(_sql(), "AI_agrosupport_cmp_layers") class ConnectionStringTests(unittest.TestCase): def test_colon_port_becomes_comma(self) -> None: text = _connection_string(_sql(server="192.168.220.50:1433")) self.assertIn("SERVER=192.168.220.50,1433;", text) def test_named_instance_unchanged(self) -> None: text = _connection_string(_sql(server=r"localhost\SQLEXPRESS")) self.assertIn(r"SERVER=localhost\SQLEXPRESS;", text) def test_comma_port_unchanged(self) -> None: text = _connection_string(_sql(server="192.168.220.50,1433")) self.assertIn("SERVER=192.168.220.50,1433;", text) def _product() -> ProductRow: return ProductRow( product_id="p1", product_name="LIETO", registration_number=None, is_organic_certified=False, systemicity="contact", has_preventive_action=True, has_curative_action=False, has_eradicant_action=False, target_crops=["grapevine"], target_diseases=["downy mildew"], active_ingredients=[], frac_codes=[], ) class QualifiedSqlTests(unittest.TestCase): def setUp(self) -> None: self.split = _sql(app_database="Agrosupport") def test_product_index_qualifies_products_and_leaves_the_catalog(self) -> None: with patch("pipeline.stages.products.fetch_all", side_effect=[[], []]) as fetch: ProductIndex.load(MagicMock(), self.split) products_sql, catalog_sql = (call.args[1] for call in fetch.call_args_list) self.assertIn("[Agrosupport].dbo.[products]", products_sql) self.assertIn("AI_agrosupport_products_catalog", catalog_sql) self.assertNotIn("Agrosupport", catalog_sql) def test_treatment_interval_qualifies_product_uses(self) -> None: with patch("pipeline.stages.products.fetch_all", return_value=[]) as fetch: last_treatment_interval_open( MagicMock(), ProductIndex([_product()]), crop_english="grapevine", disease_english="downy mildew", latest_spray=(date(2026, 4, 6), ["LIETO"]), as_of=date(2026, 4, 16), sql=self.split, ) self.assertIn("[Agrosupport].dbo.[product_uses]", fetch.call_args.args[1]) def test_incubation_coverage_qualifies_product_uses(self) -> None: as_of = date(2026, 4, 16) start = as_of - timedelta(days=2) with patch("pipeline.stages.incubations.fetch_all", return_value=[]) as fetch: incubation_start_covered( MagicMock(), ProductIndex([_product()]), crop_english="grapevine", disease_english="downy mildew", start_date=start, treatment_history=TreatmentHistory(as_of=as_of, by_day={start - timedelta(days=4): ["LIETO"]}), sql=self.split, ) self.assertIn("[Agrosupport].dbo.[product_uses]", fetch.call_args.args[1]) def test_product_json_qualifies_uses_and_chunks(self) -> None: builder = ProductJsonBuilder(ProductIndex([_product()]), "grapevine", "downy mildew", self.split) with patch("pipeline.stages.product_json.fetch_all", side_effect=[[], []]) as fetch: builder.build(MagicMock(), "LIETO") uses_sql, chunks_sql = (call.args[1] for call in fetch.call_args_list) self.assertIn("[Agrosupport].dbo.[product_uses]", uses_sql) self.assertIn("[Agrosupport].dbo.[label_chunks]", chunks_sql) def test_advice_reads_qualify_advisories(self) -> None: as_of = date(2026, 4, 16) with patch("pipeline.stages.advice_context.fetch_one", return_value=None) as fetch: load_last_advice(MagicMock(), 4012, "downy mildew", as_of, self.split) load_previous_incubation_register(MagicMock(), 4012, "downy mildew", as_of, self.split) for call in fetch.call_args_list: self.assertIn("[Agrosupport].dbo.[advisories]", call.args[1]) def test_advice_write_qualifies_advisories(self) -> None: result = AdviceResult( full_advice="text", advice_summary="summary", is_treatment_recommended=False, recommended_product_names=[], recommended_doses=[], recommended_application_date=None, ) with ( patch("pipeline.stages.advice.transaction", MagicMock()), patch("pipeline.stages.advice.execute") as execute, ): insert_advice( MagicMock(), issuance_date=date(2026, 4, 16), field_id=4012, disease_english="downy mildew", advice_input_json={}, result=result, sql=self.split, ) delete_sql, insert_sql = (call.args[1] for call in execute.call_args_list) self.assertIn("DELETE FROM [Agrosupport].dbo.[advisories]", delete_sql) self.assertIn("INSERT INTO [Agrosupport].dbo.[advisories]", insert_sql) def _called_name(node: ast.Call) -> str | None: func = node.func if isinstance(func, ast.Name): return func.id if isinstance(func, ast.Attribute): return func.attr return None def _passes_sql(node: ast.Call) -> bool: values = list(node.args) + [keyword.value for keyword in node.keywords] return any(isinstance(value, ast.Attribute) and value.attr == "sql" for value in values) class CallSiteTests(unittest.TestCase): def _calls(self, relative: str) -> list[ast.Call]: tree = ast.parse((_ROOT / relative).read_text(encoding="utf-8")) return [node for node in ast.walk(tree) if isinstance(node, ast.Call)] def test_job_passes_sql_settings(self) -> None: calls = self._calls("pipeline/job.py") expected = { "load_previous_incubation_register", "last_treatment_interval_open", "incubation_start_covered", "load_last_advice", "insert_advice", } found = {name for node in calls if (name := _called_name(node)) in expected} self.assertEqual(found, expected) for node in calls: if _called_name(node) in expected: self.assertTrue(_passes_sql(node), _called_name(node)) def test_resources_pass_sql_settings(self) -> None: calls = self._calls("pipeline/resources.py") load = [node for node in calls if _called_name(node) == "load"] builder = [node for node in calls if _called_name(node) == "ProductJsonBuilder"] self.assertEqual(len(load), 1) self.assertEqual(len(builder), 1) self.assertTrue(_passes_sql(load[0])) self.assertTrue(_passes_sql(builder[0])) def test_batch_existing_advice_uses_app_database(self) -> None: calls = self._calls("pipeline/batch.py") qualify = [node for node in calls if _called_name(node) == "qualify_app_sql"] self.assertEqual(len(qualify), 1) self.assertTrue(_passes_sql(qualify[0])) module = ast.parse((_ROOT / "pipeline/batch.py").read_text(encoding="utf-8")) constant = next( node.value.value for node in module.body if isinstance(node, ast.Assign) and isinstance(node.targets[0], ast.Name) and node.targets[0].id == "_EXISTING_ADVICE_SQL" and isinstance(node.value, ast.Constant) ) qualified = qualify_app_sql(_sql(app_database="Agrosupport"), constant) self.assertIn("FROM [Agrosupport].dbo.[advisories]", qualified) if __name__ == "__main__": unittest.main()