handoff-pro/tests/test_workflows.py

"""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()