"""Unit tests for dynamic coop, floor, and mortality discovery.""" from __future__ import annotations import tempfile import unittest from pathlib import Path from chicken_counter.config import load_or_synthesize_floor_config from chicken_counter.discovery import ( discover_all_mortality_dirs, discover_farm_structure, get_coop_for_location, resolve_floor_video_dir, sync_floor_configs, ) class TestFlexibleDiscovery(unittest.TestCase): def test_get_coop_for_location(self): self.assertEqual(get_coop_for_location("K1-L1"), "K1") self.assertEqual(get_coop_for_location("K2-L3"), "K2") self.assertEqual(get_coop_for_location("k1-l2"), "K1") self.assertEqual(get_coop_for_location("K3/L1"), "K3") self.assertEqual(get_coop_for_location("K4_L2"), "K4") self.assertEqual(get_coop_for_location("kandang-atas"), "kandang-atas") self.assertEqual(get_coop_for_location("kandang-bawah"), "kandang-bawah") self.assertEqual(get_coop_for_location("K5"), "K5") def test_discovery_on_current_workspace(self): structure = discover_farm_structure() self.assertIsNotNone(structure) # Check detected coops in workspace self.assertIn("K1", structure.coops) self.assertIn("K2", structure.coops) self.assertIn("K3", structure.coops) self.assertIn("K4", structure.coops) self.assertIn("K5", structure.coops) # Verify covered floors for K1 (3 floors) and K4 (2 floors) k1_floors = structure.get_covered_floors("K1") self.assertIn("K1-L1", k1_floors) self.assertIn("K1-L2", k1_floors) self.assertIn("K1-L3", k1_floors) k4_floors = structure.get_covered_floors("K4") self.assertIn("K4-L1", k4_floors) self.assertIn("K4-L2", k4_floors) # Verify mortality directories mort_dirs = structure.all_mortality_dirs() self.assertIn("K1", mort_dirs) self.assertIn("K2", mort_dirs) self.assertIn("K3", mort_dirs) self.assertIn("K4", mort_dirs) self.assertIn("K5", mort_dirs) self.assertEqual(mort_dirs["K1"].name, "mortality") self.assertEqual(mort_dirs["K1"].parent.name, "K1") def test_flexible_discovery_arbitrary_site(self): """Test that discovery dynamically adapts to arbitrary site layouts with varying coops and floors.""" with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) # Create arbitrary site layout: # Coop A (Kandang-Utama) has 4 floors (L1..L4) + mortality # Coop B (K9) has 1 floor (L1) + mortality # Coop C (K10) has 2 floors (L1, L2) with no mortality folder yet (tmp / "cycle1" / "Kandang-Utama" / "L1").mkdir(parents=True) (tmp / "cycle1" / "Kandang-Utama" / "L2").mkdir(parents=True) (tmp / "cycle1" / "Kandang-Utama" / "L3").mkdir(parents=True) (tmp / "cycle1" / "Kandang-Utama" / "L4").mkdir(parents=True) (tmp / "cycle1" / "Kandang-Utama" / "mortality").mkdir(parents=True) (tmp / "cycle1" / "K9" / "L1").mkdir(parents=True) (tmp / "cycle1" / "K9" / "mortality").mkdir(parents=True) (tmp / "cycle1" / "K10" / "L1").mkdir(parents=True) (tmp / "cycle1" / "K10" / "L2").mkdir(parents=True) structure = discover_farm_structure(videos_root=tmp) self.assertTrue("KANDANG-UTAMA" in structure.coops or "Kandang-Utama" in structure.coops) coop_u = structure.get_coop("Kandang-Utama") self.assertIsNotNone(coop_u) self.assertEqual(len(coop_u.covered_floors), 4) self.assertIsNotNone(coop_u.mortality_dir) coop_k9 = structure.get_coop("K9") self.assertIsNotNone(coop_k9) self.assertEqual(len(coop_k9.covered_floors), 1) self.assertIsNotNone(coop_k9.mortality_dir) coop_k10 = structure.get_coop("K10") self.assertIsNotNone(coop_k10) self.assertEqual(len(coop_k10.covered_floors), 2) self.assertIsNone(coop_k10.mortality_dir) m_dirs = structure.all_mortality_dirs() self.assertIn("K9", m_dirs) self.assertTrue("Kandang-Utama" in m_dirs or "KANDANG-UTAMA" in m_dirs) self.assertNotIn("K10", m_dirs) def test_resolve_floor_video_dir(self): with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) # Create hierarchical directory hier_dir = tmp / "cycle7" / "K1" / "L1" hier_dir.mkdir(parents=True) # Query using old flat path flat_query = tmp / "cycle7" / "K1-L1" resolved = resolve_floor_video_dir(flat_query, "K1-L1") self.assertEqual(resolved, hier_dir) def test_sync_floor_configs(self): """Test auto-generating floor YAML configs for newly discovered coops/floors.""" with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) vid_dir = tmp / "VIDEOS" / "cycle7" cfg_dir = tmp / "configs" / "floor_config" # Create new coop K8 with 3 floors (vid_dir / "K8" / "L1").mkdir(parents=True) (vid_dir / "K8" / "L2").mkdir(parents=True) (vid_dir / "K8" / "L3").mkdir(parents=True) created = sync_floor_configs(videos_root=vid_dir.parent, configs_dir=cfg_dir) self.assertEqual(len(created), 3) # Check that files were written self.assertTrue((cfg_dir / "K8-L1.yaml").exists()) self.assertTrue((cfg_dir / "K8-L2.yaml").exists()) self.assertTrue((cfg_dir / "K8-L3.yaml").exists()) content = (cfg_dir / "K8-L1.yaml").read_text() self.assertIn('location: "K8-L1"', content) self.assertIn('root_dir: "../VIDEOS/cycle7/K8/L1"', content) # Running sync again should create 0 files created_again = sync_floor_configs(videos_root=vid_dir.parent, configs_dir=cfg_dir) self.assertEqual(len(created_again), 0) def test_load_or_synthesize_floor_config(self): """Test virtual in-memory config synthesis without any YAML file.""" # Load virtual floor K9-L4 (no YAML exists) settings = load_or_synthesize_floor_config("K9-L4") self.assertEqual(settings.batch.location, "K9-L4") self.assertTrue(settings.batch.root_dir.endswith("VIDEOS/cycle7/K9/L4")) self.assertEqual(len(settings.cameras), 4) self.assertIn("CC1", settings.cameras) def test_dated_mortality_discovery(self): """Test that dated mortality subfolders (e.g. K1/mortality/2026-08-20/) are discovered.""" with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) mort_day = tmp / "cycle7" / "K1" / "mortality" / "2026-08-20" mort_day.mkdir(parents=True) report = mort_day / "mortality_report.json" report.write_text('{"date":"2026-08-20","total_mortality_count":15,"results":[]}') structure = discover_farm_structure(videos_root=tmp) self.assertIn("K1", structure.coops) self.assertIsNotNone(structure.coops["K1"].mortality_dir) def test_stationary_camera_config(self): """Test retrieving stationary camera sources for farm coops.""" from chicken_counter.mortality import capture_camera_snapshot, get_camera_source_for_coop cam_k1 = get_camera_source_for_coop("K1") self.assertIsNotNone(cam_k1) self.assertTrue("192.168.1.101" in str(cam_k1) or "rtsp" in str(cam_k1)) cam_k5 = get_camera_source_for_coop("K5") self.assertIsNotNone(cam_k5) # Non-existent camera stream capture should return None gracefully without raising invalid_cap = capture_camera_snapshot("rtsp://invalid-non-existent-host.local:554/live", timeout_sec=0.5, warmup_frames=1) self.assertIsNone(invalid_cap) if __name__ == "__main__": unittest.main()