Initial commit: Handoff Pro MCP server for Kellow Construction

tests/test_workflows.py

@@ -0,0 +1,149 @@
+"""End-to-end workflow tests for Handoff Pro."""
+import json, os, sys, tempfile, unittest
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from lib import db as db_mod
+from scripts import pricing_catalog, project_mgmt, estimate_generator
+from scripts import proposal_drafter, invoice_mgmt, daily_log
+from scripts import scope_generator, material_list, budget_tracker
+from scripts import project_scheduler, signature_mgmt, activity_logger, site_analyzer
+
+def fresh_db():
+    """Point to a fresh temp DB and initialize it."""
+    import tempfile as tf
+    path = tf.mktemp(suffix=".db")
+    db_mod.DB_PATH = path
+    db_mod.init_db()
+    return path
+
+class TestWorkflow1_NewEstimate(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+        pricing_catalog.seed_default()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    def test_full_estimate_workflow(self):
+        # Generate estimate from description
+        result = estimate_generator.generate("kitchen remodel, 200sqft, new cabinets, countertops, flooring")
+        self.assertIn("id", result)
+        self.assertGreater(result["total_cost"], 0)
+        # Verify estimate retrievable
+        est = project_mgmt.get_estimate(result["id"])
+        self.assertEqual(est["id"], result["id"])
+        self.assertGreater(len(est["line_items"]), 0)
+
+class TestWorkflow2_Proposal(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+        pricing_catalog.seed_default()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    def test_proposal_generation(self):
+        est = estimate_generator.generate("bathroom renovation, tile, plumbing, 100sqft")
+        result = proposal_drafter.draft(est["id"])
+        self.assertIn("proposal", result)
+        self.assertIn("Kellow Construction", result["proposal"])
+        self.assertIn("Payment Terms", result["proposal"])
+
+class TestWorkflow3_DailyLog(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    def test_daily_log_workflow(self):
+        job = project_mgmt.create_job("Test Client", "123 Main St")
+        log = daily_log.create(job["id"], "Framed exterior walls", crew_names="Mike, Sarah", hours_worked=8)
+        self.assertIn("id", log)
+        logs = daily_log.list_logs(job["id"])
+        self.assertEqual(len(logs["logs"]), 1)
+
+class TestWorkflow4_Invoice(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+        pricing_catalog.seed_default()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    def test_invoice_lifecycle(self):
+        est = estimate_generator.generate("deck build, 300sqft")
+        inv = invoice_mgmt.create(est["id"])
+        self.assertIn("invoice_number", inv)
+        self.assertGreater(inv["amount_due"], 0)
+        # Mark paid
+        paid = invoice_mgmt.mark_paid(inv["id"])
+        self.assertEqual(paid["status"], "paid")
+
+class TestWorkflow5_JobTreadSync(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    @patch("scripts.jobtread_sync.pave_query")
+    @patch("scripts.jobtread_sync.get_org_id", return_value="test_org")
+    def test_pull_jobs(self, mock_org, mock_pave):
+        mock_pave.return_value = {"organization": {"jobs": {"nodes": [
+            {"id": "jt123", "name": "Test Job", "status": "active", "createdAt": "2026-01-01",
+             "account": {"id": "acc1", "name": "John Smith"}, "location": {"address": "456 Oak Ave"}}
+        ]}}}
+        from scripts import jobtread_sync
+        result = jobtread_sync.pull_jobs()
+        self.assertEqual(result["synced"], 1)
+        # Verify in local DB
+        jobs = project_mgmt.list_jobs()
+        self.assertGreater(len(jobs["jobs"]), 0)
+
+class TestBudgetTracker(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+        pricing_catalog.seed_default()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    def test_budget_report(self):
+        est = estimate_generator.generate("framing project, 500sqft")
+        job_id = project_mgmt.get_estimate(est["id"])["job_id"]
+        daily_log.create(job_id, "Day 1 framing", hours_worked=10)
+        report = budget_tracker.report(job_id)
+        self.assertIn("budget", report)
+        self.assertIn("actual", report)
+        self.assertEqual(report["actual"]["hours_worked"], 10)
+
+class TestSiteAnalyzer(unittest.TestCase):
+    def test_issue_detection(self):
+        result = site_analyzer.analyze("Found water damage in the corner near the window, also some cracking in foundation")
+        self.assertGreater(result["issues_found"], 0)
+        self.assertGreater(result["total_cost_impact"], 0)
+
+class TestScopeAndMaterials(unittest.TestCase):
+    def setUp(self):
+        self.db = fresh_db()
+        pricing_catalog.seed_default()
+
+    def tearDown(self):
+        os.unlink(self.db)
+
+    def test_scope_generation(self):
+        est = estimate_generator.generate("bathroom tile, plumbing fixtures")
+        scope = scope_generator.generate(est["id"])
+        self.assertIn("scope", scope)
+        self.assertIn("Scope of Work", scope["scope"])
+
+    def test_material_list(self):
+        est = estimate_generator.generate("kitchen cabinets, countertops, flooring")
+        mats = material_list.generate(est["id"])
+        self.assertGreater(mats["items_count"], 0)
+
+if __name__ == "__main__":
+    unittest.main()