#!/usr/bin/env python3 """ Unit tests for session-log-git-sync.py and migrate-legacy-session-logs.py

Tests: J.27.5 Track: J.27 - Project-Scoped Session Log Sync Author: Claude (Opus 4.6) """

import json import os import stat import sys import tempfile import unittest from collections import Counter from pathlib import Path from unittest.mock import MagicMock, patch

Add scripts directory to path

sys.path.insert(0, str(Path(file).parent.parent))

Load session-log-git-sync.py as module (hyphenated filename)

from importlib.util import module_from_spec, spec_from_file_location

sync_spec = spec_from_file_location( "session_log_git_sync", Path(file).parent.parent / "session-log-git-sync.py" ) sync_mod = module_from_spec(sync_spec) sync_spec.loader.exec_module(sync_mod)

migrate_spec = spec_from_file_location( "migrate_legacy_session_logs", Path(file).parent.parent / "migrate-legacy-session-logs.py" ) migrate_mod = module_from_spec(migrate_spec) migrate_spec.loader.exec_module(migrate_mod)

=============================================================================

Tests for session-log-git-sync.py

=============================================================================

class TestIsCustomerProject(unittest.TestCase): """J.27.5.1: Test customer project detection."""

def test_cust_prefix(self):
    self.assertTrue(sync_mod.is_customer_project("CUST-avivatec-fpa"))

def test_cust_prefix_short(self):
    self.assertTrue(sync_mod.is_customer_project("CUST-x"))

def test_pilot(self):
    self.assertFalse(sync_mod.is_customer_project("PILOT"))

def test_internal(self):
    self.assertFalse(sync_mod.is_customer_project("INTERNAL"))

def test_empty(self):
    self.assertFalse(sync_mod.is_customer_project(""))

class TestGetTenantFromProject(unittest.TestCase): """J.27.5.1: Test tenant extraction from project ID."""

def test_avivatec(self):
    self.assertEqual(sync_mod.get_tenant_from_project("CUST-avivatec-fpa"), "avivatec")

def test_other_tenant(self):
    self.assertEqual(sync_mod.get_tenant_from_project("CUST-acme-prod"), "acme")

def test_not_customer(self):
    self.assertIsNone(sync_mod.get_tenant_from_project("PILOT"))

def test_empty(self):
    self.assertIsNone(sync_mod.get_tenant_from_project(""))

class TestProjectsForTenant(unittest.TestCase): """J.27.5.1: Test tenant → project resolution."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()
    self.projects_dir = Path(self.tmpdir) / "projects"
    self.projects_dir.mkdir()
    (self.projects_dir / "PILOT").mkdir()
    (self.projects_dir / "CUST-avivatec-fpa").mkdir()
    (self.projects_dir / "CUST-avivatec-staging").mkdir()
    (self.projects_dir / "CUST-acme-prod").mkdir()

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

def test_avivatec_matches(self):
    result = sync_mod.projects_for_tenant("avivatec", self.projects_dir)
    self.assertEqual(result, ["CUST-avivatec-fpa", "CUST-avivatec-staging"])

def test_acme_matches(self):
    result = sync_mod.projects_for_tenant("acme", self.projects_dir)
    self.assertEqual(result, ["CUST-acme-prod"])

def test_unknown_tenant(self):
    result = sync_mod.projects_for_tenant("unknown", self.projects_dir)
    self.assertEqual(result, [])

def test_nonexistent_dir(self):
    result = sync_mod.projects_for_tenant("avivatec", Path("/nonexistent"))
    self.assertEqual(result, [])

class TestEnforceCustomerPermissions(unittest.TestCase): """J.27.5.1: Test customer directory permission enforcement."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()
    self.project_dir = Path(self.tmpdir) / "CUST-avivatec-fpa"
    self.project_dir.mkdir()
    # Set to world-readable initially
    self.project_dir.chmod(0o755)

def tearDown(self):
    # Restore permissions for cleanup
    self.project_dir.chmod(0o755)
    import shutil
    shutil.rmtree(self.tmpdir)

def test_sets_700_on_customer(self):
    sync_mod.enforce_customer_permissions(self.project_dir, "CUST-avivatec-fpa")
    mode = self.project_dir.stat().st_mode & 0o777
    self.assertEqual(mode, 0o700)

def test_skips_non_customer(self):
    pilot_dir = Path(self.tmpdir) / "PILOT"
    pilot_dir.mkdir()
    pilot_dir.chmod(0o755)
    sync_mod.enforce_customer_permissions(pilot_dir, "PILOT")
    mode = pilot_dir.stat().st_mode & 0o777
    self.assertEqual(mode, 0o755)

def test_skips_nonexistent(self):
    # Should not raise
    sync_mod.enforce_customer_permissions(Path("/nonexistent"), "CUST-x")

def test_already_700(self):
    self.project_dir.chmod(0o700)
    # Should not raise, should be idempotent
    sync_mod.enforce_customer_permissions(self.project_dir, "CUST-avivatec-fpa")
    mode = self.project_dir.stat().st_mode & 0o777
    self.assertEqual(mode, 0o700)

class TestCheckPiiInFile(unittest.TestCase): """J.27.5.1: Test PII detection in files."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

def _write_file(self, name, content):
    path = Path(self.tmpdir) / name
    path.write_text(content)
    return path

def test_email_detected(self):
    f = self._write_file("test.md", "Contact john.doe@example.com for details")
    findings = sync_mod.check_pii_in_file(f)
    self.assertEqual(len(findings), 1)
    self.assertEqual(findings[0][1], "email")
    self.assertIn("john.doe@example.com", findings[0][2])

def test_ssn_detected(self):
    f = self._write_file("test.md", "SSN: 123-45-6789")
    findings = sync_mod.check_pii_in_file(f)
    # Should find SSN-like pattern
    ssn_findings = [f for f in findings if f[1] == "ssn"]
    self.assertTrue(len(ssn_findings) >= 1)

def test_phone_detected(self):
    f = self._write_file("test.md", "Call (555) 123-4567 for support")
    findings = sync_mod.check_pii_in_file(f)
    phone_findings = [f for f in findings if f[1] == "phone"]
    self.assertTrue(len(phone_findings) >= 1)

def test_allowlisted_email(self):
    f = self._write_file("test.md", "Co-Authored-By: noreply@anthropic.com")
    findings = sync_mod.check_pii_in_file(f)
    email_findings = [f for f in findings if f[1] == "email"]
    self.assertEqual(len(email_findings), 0, "noreply@anthropic.com should be allowlisted")

def test_task_id_not_ssn(self):
    f = self._write_file("test.md", "Task A.9.1.1: Complete migration")
    findings = sync_mod.check_pii_in_file(f)
    # Task IDs should be allowlisted
    ssn_findings = [f for f in findings if f[1] == "ssn"]
    self.assertEqual(len(ssn_findings), 0, "Task IDs should be allowlisted")

def test_clean_file(self):
    f = self._write_file("test.md", "# Session Log\n\nDid some coding today.")
    findings = sync_mod.check_pii_in_file(f)
    self.assertEqual(len(findings), 0)

def test_nonexistent_file(self):
    findings = sync_mod.check_pii_in_file(Path("/nonexistent/file.md"))
    self.assertEqual(len(findings), 0)

class TestCheckPiiRedaction(unittest.TestCase): """J.27.5.1: Test bulk PII scanning."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

def test_only_scans_customer_files(self):
    # Create files
    pilot_file = Path(self.tmpdir) / "pilot.md"
    pilot_file.write_text("Contact john@example.com")
    cust_file = Path(self.tmpdir) / "cust.md"
    cust_file.write_text("Contact jane@example.com")

    logs = [
        (pilot_file, "projects/PILOT/SESSION-LOG-2026-01-01.md"),
        (cust_file, "projects/CUST-avivatec-fpa/SESSION-LOG-2026-01-01.md"),
    ]
    result = sync_mod.check_pii_redaction(logs)
    # Only the CUST file should be flagged
    self.assertEqual(len(result), 1)
    self.assertEqual(result[0][0], cust_file)

def test_clean_customer_files(self):
    cust_file = Path(self.tmpdir) / "clean.md"
    cust_file.write_text("# Session Log\nDid work today.")
    logs = [
        (cust_file, "projects/CUST-avivatec-fpa/SESSION-LOG-2026-01-01.md"),
    ]
    result = sync_mod.check_pii_redaction(logs)
    self.assertEqual(len(result), 0)

class TestDiscoverAllLogs(unittest.TestCase): """J.27.5.1-5.5: Test project-scoped log discovery."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()
    self.logs_dir = Path(self.tmpdir) / "session-logs"
    self.logs_dir.mkdir()
    projects = self.logs_dir / "projects"
    projects.mkdir()

    # PILOT project
    uuid = "d3d3a316-09c6-8f41-4a3f-d93e422d199c"
    pilot_dir = projects / "PILOT" / uuid
    pilot_dir.mkdir(parents=True)
    (pilot_dir / "SESSION-LOG-2026-01-01.md").write_text("PILOT log 1")
    (pilot_dir / "SESSION-LOG-2026-01-02.md").write_text("PILOT log 2")

    # Customer project
    cust_dir = projects / "CUST-avivatec-fpa" / uuid
    cust_dir.mkdir(parents=True)
    (cust_dir / "SESSION-LOG-2026-01-01.md").write_text("CUST log 1")

    # Symlinks at root (should be skipped by discover - they're legacy)
    self.flat_link = self.logs_dir / "SESSION-LOG-2026-01-01.md"
    if not self.flat_link.exists():
        self.flat_link.symlink_to(
            pilot_dir / "SESSION-LOG-2026-01-01.md"
        )

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

@patch.object(sync_mod, "SESSION_LOGS_DIR")
def test_discovers_all_projects(self, mock_dir):
    mock_dir.__class__ = type(self.logs_dir)
    with patch.object(sync_mod, "SESSION_LOGS_DIR", self.logs_dir):
        results = sync_mod.discover_all_logs()
    self.assertEqual(len(results), 3)

@patch.object(sync_mod, "SESSION_LOGS_DIR")
def test_project_filter(self, mock_dir):
    with patch.object(sync_mod, "SESSION_LOGS_DIR", self.logs_dir):
        results = sync_mod.discover_all_logs(project_filter="PILOT")
    self.assertEqual(len(results), 2)
    for _, dest in results:
        self.assertIn("PILOT", dest)

@patch.object(sync_mod, "SESSION_LOGS_DIR")
def test_tenant_filter(self, mock_dir):
    with patch.object(sync_mod, "SESSION_LOGS_DIR", self.logs_dir):
        results = sync_mod.discover_all_logs(tenant_filter="avivatec")
    self.assertEqual(len(results), 1)
    self.assertIn("CUST-avivatec-fpa", results[0][1])

@patch.object(sync_mod, "SESSION_LOGS_DIR")
def test_unknown_tenant_empty(self, mock_dir):
    with patch.object(sync_mod, "SESSION_LOGS_DIR", self.logs_dir):
        results = sync_mod.discover_all_logs(tenant_filter="unknown")
    self.assertEqual(len(results), 0)

=============================================================================

Tests for migrate-legacy-session-logs.py

=============================================================================

class TestParseTaskIds(unittest.TestCase): """J.27.5.1: Test task ID extraction."""

def test_standard_tracks(self):
    content = "Working on A.9.1.1 and B.3.2 today. Also H.8.1.6 stuff."
    result = migrate_mod.parse_task_ids(content)
    self.assertEqual(result["A"], 1)
    self.assertEqual(result["B"], 1)
    self.assertEqual(result["H"], 1)

def test_fpa_prefix(self):
    content = "FPA.1.2.3 customer work, also FPA.2.1 billing."
    result = migrate_mod.parse_task_ids(content)
    self.assertEqual(result["FPA"], 2)

def test_no_task_ids(self):
    content = "Just a regular day with no tasks."
    result = migrate_mod.parse_task_ids(content)
    self.assertEqual(len(result), 0)

def test_multiple_same_track(self):
    content = "A.1.1, A.1.2, A.2.1, A.9.1.1"
    result = migrate_mod.parse_task_ids(content)
    self.assertEqual(result["A"], 4)

def test_mixed_tracks(self):
    content = "A.1.1 and FPA.2.1 and C.3.1"
    result = migrate_mod.parse_task_ids(content)
    self.assertEqual(result["A"], 1)
    self.assertEqual(result["FPA"], 1)
    self.assertEqual(result["C"], 1)

class TestParsePathHints(unittest.TestCase): """J.27.5.1: Test path hint detection."""

def test_rollout_master(self):
    content = "Working in /Users/hal/PROJECTS/coditect-rollout-master/scripts"
    result = migrate_mod.parse_path_hints(content)
    self.assertEqual(result["PILOT"], 1)

def test_avivatec(self):
    content = "Modified /PROJECTS/coditect-jv-avivatec/backend/api.py"
    result = migrate_mod.parse_path_hints(content)
    self.assertEqual(result["CUST-avivatec-fpa"], 1)

def test_both(self):
    content = (
        "Started in coditect-rollout-master then switched to "
        "coditect-jv-avivatec for customer work."
    )
    result = migrate_mod.parse_path_hints(content)
    self.assertEqual(result["PILOT"], 1)
    self.assertEqual(result["CUST-avivatec-fpa"], 1)

def test_no_hints(self):
    content = "Just coding at home."
    result = migrate_mod.parse_path_hints(content)
    self.assertEqual(len(result), 0)

def test_multiple_occurrences(self):
    content = (
        "coditect-rollout-master/scripts and "
        "coditect-rollout-master/docs and "
        "coditect-rollout-master/CLAUDE.md"
    )
    result = migrate_mod.parse_path_hints(content)
    self.assertEqual(result["PILOT"], 3)

class TestAttributeLog(unittest.TestCase): """J.27.5.1: Test log attribution scoring."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

def _write(self, name, content):
    path = Path(self.tmpdir) / name
    path.write_text(content)
    return path

def test_pilot_by_task_ids(self):
    f = self._write("log.md", "Working on A.9.1.1, B.3.2, C.5.1")
    result = migrate_mod.attribute_log(f)
    self.assertEqual(result, "PILOT")

def test_avivatec_by_task_ids(self):
    f = self._write("log.md", "FPA.1.2.3, FPA.2.1, FPA.3.1.1")
    result = migrate_mod.attribute_log(f)
    self.assertEqual(result, "CUST-avivatec-fpa")

def test_pilot_by_path_hints(self):
    f = self._write("log.md", "All work in coditect-rollout-master today")
    result = migrate_mod.attribute_log(f)
    self.assertEqual(result, "PILOT")

def test_mixed_with_pilot_dominant(self):
    f = self._write("log.md", (
        "A.1.1, A.2.1, A.3.1, B.1.1, C.1.1\n"
        "FPA.1.1\n"
        "coditect-rollout-master/scripts"
    ))
    result = migrate_mod.attribute_log(f)
    self.assertEqual(result, "PILOT")

def test_empty_content(self):
    f = self._write("log.md", "")
    result = migrate_mod.attribute_log(f)
    self.assertIsNone(result)

def test_nonexistent_file(self):
    result = migrate_mod.attribute_log(Path("/nonexistent/log.md"))
    self.assertIsNone(result)

class TestFindLegacyLogs(unittest.TestCase): """J.27.5.5: Test backward compatibility with legacy flat logs."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()
    self.logs_dir = Path(self.tmpdir) / "session-logs"
    self.logs_dir.mkdir()

    # Real legacy flat files
    (self.logs_dir / "SESSION-LOG-2026-01-01.md").write_text("Legacy log 1")
    (self.logs_dir / "SESSION-LOG-2026-01-02.md").write_text("Legacy log 2")

    # Symlink (should be skipped)
    target = self.logs_dir / "SESSION-LOG-2026-01-01.md"
    link = self.logs_dir / "SESSION-LOG-2026-01-03.md"
    link.symlink_to(target)

    # Non-matching file
    (self.logs_dir / "README.md").write_text("Not a log")

    # Project dir (logs inside should be skipped)
    proj_dir = self.logs_dir / "projects" / "PILOT"
    proj_dir.mkdir(parents=True)
    (proj_dir / "SESSION-LOG-2026-01-01.md").write_text("Project log")

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

def test_finds_real_files_only(self):
    result = migrate_mod.find_legacy_logs(self.logs_dir)
    names = [f.name for f in result]
    self.assertIn("SESSION-LOG-2026-01-01.md", names)
    self.assertIn("SESSION-LOG-2026-01-02.md", names)
    # Symlink should be skipped
    self.assertNotIn("SESSION-LOG-2026-01-03.md", names)
    self.assertEqual(len(result), 2)

class TestMigrateLogs(unittest.TestCase): """J.27.5.3: Test multi-project migration dry-run."""

def setUp(self):
    self.tmpdir = tempfile.mkdtemp()
    self.logs_dir = Path(self.tmpdir) / "session-logs"
    self.logs_dir.mkdir()

    # Create a legacy log with PILOT content
    (self.logs_dir / "SESSION-LOG-2026-01-15.md").write_text(
        "# Session Log 2026-01-15\n\n"
        "Working on A.9.1.1 in coditect-rollout-master/scripts today.\n"
        "Also did B.3.2 and C.5.1.\n"
    )

    # Create machine-id.json
    machine_id = Path(self.tmpdir) / "machine-id.json"
    machine_id.write_text(json.dumps({"machine_uuid": "test-uuid-1234"}))

def tearDown(self):
    import shutil
    shutil.rmtree(self.tmpdir)

@patch.object(migrate_mod, "get_session_logs_dir")
@patch.object(migrate_mod, "get_machine_uuid")
def test_dry_run(self, mock_uuid, mock_dir):
    mock_dir.return_value = self.logs_dir
    mock_uuid.return_value = "test-uuid-1234"

    migrated, skipped, errors = migrate_mod.migrate_logs(
        dry_run=True, verbose=False
    )
    self.assertEqual(migrated, 1)
    self.assertEqual(skipped, 0)
    self.assertEqual(errors, 0)

@patch.object(migrate_mod, "get_session_logs_dir")
@patch.object(migrate_mod, "get_machine_uuid")
def test_actual_migration(self, mock_uuid, mock_dir):
    mock_dir.return_value = self.logs_dir
    mock_uuid.return_value = "test-uuid-1234"

    migrated, skipped, errors = migrate_mod.migrate_logs(
        dry_run=False, verbose=False
    )
    self.assertEqual(migrated, 1)
    self.assertEqual(errors, 0)

    # Verify file was moved
    dest = self.logs_dir / "projects" / "PILOT" / "test-uuid-1234" / "SESSION-LOG-2026-01-15.md"
    self.assertTrue(dest.exists())

    # Verify source was removed
    src = self.logs_dir / "SESSION-LOG-2026-01-15.md"
    self.assertFalse(src.exists())

@patch.object(migrate_mod, "get_session_logs_dir")
@patch.object(migrate_mod, "get_machine_uuid")
def test_skip_existing(self, mock_uuid, mock_dir):
    mock_dir.return_value = self.logs_dir
    mock_uuid.return_value = "test-uuid-1234"

    # Pre-create destination
    dest_dir = self.logs_dir / "projects" / "PILOT" / "test-uuid-1234"
    dest_dir.mkdir(parents=True)
    (dest_dir / "SESSION-LOG-2026-01-15.md").write_text("Already here")

    migrated, skipped, errors = migrate_mod.migrate_logs(
        dry_run=False, verbose=False
    )
    self.assertEqual(migrated, 0)
    self.assertEqual(skipped, 1)
    self.assertEqual(errors, 0)

if name == "main": unittest.main()