auto-claude: subtask-7-2 - Test collaboration matching

Created comprehensive test suite for IT audit collaboration matching:

1. Unit tests (tests/test_it_audit_collaboration.py):
   - 12 tests verifying all 6 match types
   - Backup replication, shared licensing, Teams federation
   - Shared monitoring, collective purchasing, knowledge sharing
   - Edge cases for size parsing and similarity

2. Integration test script (scripts/test_collaboration_matching.py):
   - Creates test audits with matching criteria
   - Runs collaboration matching algorithm
   - Verifies matches saved to database

All unit tests pass (12/12).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

scripts/test_collaboration_matching.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""
+Test Collaboration Matching Integration
+========================================
+
+This script tests the collaboration matching functionality by:
+1. Creating two IT audits with matching criteria (Proxmox PBS + open_to_backup_replication)
+2. Running the collaboration matching algorithm
+3. Verifying matches appear in the database
+
+Usage:
+    # From the worktree directory with DEV database running:
+    DATABASE_URL=postgresql://nordabiz_app:NordaBiz2025Secure@localhost:5433/nordabiz \
+        python3 scripts/test_collaboration_matching.py
+
+Requirements:
+    - PostgreSQL DEV database running on localhost:5433
+    - IT audit tables created (run migration first)
+
+Author: Norda Biznes Development Team
+Created: 2026-01-09
+"""
+
+import os
+import sys
+from datetime import datetime
+
+# Set database URL for DEV environment
+if 'DATABASE_URL' not in os.environ:
+    os.environ['DATABASE_URL'] = 'postgresql://nordabiz_app:NordaBiz2025Secure@localhost:5433/nordabiz'
+
+# Import after setting DATABASE_URL
+from database import SessionLocal, Company, ITAudit, ITCollaborationMatch
+from it_audit_service import ITAuditService
+
+
+def get_test_companies(db, limit=2):
+    """Get two test companies from the database"""
+    companies = db.query(Company).filter(
+        Company.status == 'active'
+    ).limit(limit).all()
+    return companies
+
+
+def create_test_audit(db, company_id: int, audit_data: dict) -> ITAudit:
+    """Create a test IT audit for a company"""
+    service = ITAuditService(db)
+    return service.save_audit(company_id, audit_data)
+
+
+def run_matching_test():
+    """Run the collaboration matching test"""
+    print("=" * 60)
+    print("IT Audit Collaboration Matching Test")
+    print("=" * 60)
+
+    db = SessionLocal()
+    try:
+        # Step 1: Get two test companies
+        print("\n[1] Getting test companies...")
+        companies = get_test_companies(db, limit=2)
+
+        if len(companies) < 2:
+            print("ERROR: Need at least 2 active companies in database")
+            return False
+
+        company_a, company_b = companies[0], companies[1]
+        print(f"    Company A: {company_a.name} (ID: {company_a.id})")
+        print(f"    Company B: {company_b.name} (ID: {company_b.id})")
+
+        # Step 2: Check for existing audits and delete them for clean test
+        print("\n[2] Cleaning up existing test audits...")
+        db.query(ITAudit).filter(
+            ITAudit.company_id.in_([company_a.id, company_b.id])
+        ).delete(synchronize_session='fetch')
+
+        db.query(ITCollaborationMatch).filter(
+            (ITCollaborationMatch.company_a_id.in_([company_a.id, company_b.id])) |
+            (ITCollaborationMatch.company_b_id.in_([company_a.id, company_b.id]))
+        ).delete(synchronize_session='fetch')
+
+        db.commit()
+        print("    Existing audits and matches cleaned up.")
+
+        # Step 3: Create audit for Company A with PBS + open_to_backup_replication
+        print("\n[3] Creating IT audit for Company A with Proxmox PBS...")
+        audit_a_data = {
+            'has_proxmox_pbs': True,
+            'open_to_backup_replication': True,
+            'backup_solution': 'Proxmox Backup Server',
+            'backup_frequency': 'daily',
+            'has_edr': True,
+            'has_mfa': True,
+            'has_azure_ad': True,
+            'has_m365': True,
+            'open_to_shared_licensing': True,
+            'open_to_teams_federation': True,
+            'employee_count': '11-50',
+            'monitoring_solution': 'Zabbix',
+            'open_to_shared_monitoring': True,
+        }
+        audit_a = create_test_audit(db, company_a.id, audit_a_data)
+        print(f"    Audit A created: ID={audit_a.id}, PBS={audit_a.has_proxmox_pbs}, "
+              f"Open to backup replication={audit_a.open_to_backup_replication}")
+        print(f"    Scores: overall={audit_a.overall_score}, security={audit_a.security_score}, "
+              f"collaboration={audit_a.collaboration_score}")
+
+        # Step 4: Create audit for Company B with PBS + open_to_backup_replication
+        print("\n[4] Creating IT audit for Company B with Proxmox PBS...")
+        audit_b_data = {
+            'has_proxmox_pbs': True,
+            'open_to_backup_replication': True,
+            'backup_solution': 'Proxmox Backup Server',
+            'backup_frequency': 'daily',
+            'has_edr': True,
+            'has_mfa': True,
+            'has_azure_ad': True,
+            'has_m365': True,
+            'open_to_shared_licensing': True,
+            'open_to_teams_federation': True,
+            'employee_count': '51-100',
+            'monitoring_solution': 'Zabbix',
+            'open_to_shared_monitoring': True,
+        }
+        audit_b = create_test_audit(db, company_b.id, audit_b_data)
+        print(f"    Audit B created: ID={audit_b.id}, PBS={audit_b.has_proxmox_pbs}, "
+              f"Open to backup replication={audit_b.open_to_backup_replication}")
+        print(f"    Scores: overall={audit_b.overall_score}, security={audit_b.security_score}, "
+              f"collaboration={audit_b.collaboration_score}")
+
+        # Step 5: Run collaboration matching for Company A
+        print("\n[5] Running collaboration matching for Company A...")
+        service = ITAuditService(db)
+        matches = service.find_collaboration_matches(company_a.id)
+        print(f"    Found {len(matches)} potential matches")
+
+        # Step 6: Save matches to database
+        print("\n[6] Saving matches to database...")
+        for match in matches:
+            saved_match = service.save_collaboration_match(match)
+            print(f"    Saved: {match.company_a_name} <-> {match.company_b_name}")
+            print(f"           Type: {match.match_type}, Score: {match.match_score}")
+            print(f"           Reason: {match.match_reason}")
+
+        # Step 7: Verify matches in database
+        print("\n[7] Verifying matches in database...")
+        db_matches = db.query(ITCollaborationMatch).filter(
+            (ITCollaborationMatch.company_a_id == company_a.id) |
+            (ITCollaborationMatch.company_b_id == company_a.id)
+        ).all()
+
+        print(f"    Total matches in DB: {len(db_matches)}")
+
+        # Check for backup_replication match
+        backup_matches = [m for m in db_matches if m.match_type == 'backup_replication']
+        has_backup_match = len(backup_matches) > 0
+
+        if has_backup_match:
+            print("\n✅ SUCCESS: Backup replication match found!")
+            bm = backup_matches[0]
+            print(f"    Company A ID: {bm.company_a_id}")
+            print(f"    Company B ID: {bm.company_b_id}")
+            print(f"    Match Score: {bm.match_score}")
+            print(f"    Status: {bm.status}")
+        else:
+            print("\n❌ FAILED: No backup replication match found!")
+
+        # Print all match types found
+        print("\n[8] All match types found:")
+        for match in db_matches:
+            print(f"    - {match.match_type}: {match.company_a_id} <-> {match.company_b_id} (score: {match.match_score})")
+
+        # Summary
+        print("\n" + "=" * 60)
+        print("TEST SUMMARY")
+        print("=" * 60)
+        print(f"Companies tested: {company_a.name}, {company_b.name}")
+        print(f"Audits created: 2")
+        print(f"Matches found: {len(db_matches)}")
+        print(f"Backup replication match: {'YES ✅' if has_backup_match else 'NO ❌'}")
+
+        # Expected matches based on test data
+        expected_match_types = ['backup_replication', 'shared_licensing', 'teams_federation', 'shared_monitoring']
+        found_types = {m.match_type for m in db_matches}
+        missing_types = set(expected_match_types) - found_types
+
+        if missing_types:
+            print(f"\nMissing expected match types: {missing_types}")
+
+        return has_backup_match
+
+    except Exception as e:
+        print(f"\n❌ ERROR: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+    finally:
+        db.close()
+
+
+if __name__ == '__main__':
+    success = run_matching_test()
+    sys.exit(0 if success else 1)

tests/test_it_audit_collaboration.py

@@ -0,0 +1,394 @@
+"""
+Test IT Audit Collaboration Matching
+====================================
+
+Tests for the collaboration matching functionality in IT Audit Service.
+
+Test cases:
+1. Backup replication matching (Proxmox PBS)
+2. Shared licensing matching (M365)
+3. Teams federation matching (Azure AD)
+4. Shared monitoring matching (Zabbix)
+5. Collective purchasing matching (similar size)
+6. Knowledge sharing matching (similar tech stack)
+
+Author: Norda Biznes Development Team
+Created: 2026-01-09
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+from dataclasses import dataclass
+
+# Add parent directory (worktree root) to path for imports
+WORKTREE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, WORKTREE_ROOT)
+
+# Mock database module before importing it_audit_service
+# This prevents database connection errors during import
+sys.modules['database'] = MagicMock()
+
+# Now we can safely import it_audit_service
+from it_audit_service import ITAuditService, CollaborationMatch
+
+
+class MockCompany:
+    """Mock Company object for testing"""
+    def __init__(self, id: int, name: str, slug: str):
+        self.id = id
+        self.name = name
+        self.slug = slug
+
+
+class MockITAudit:
+    """Mock ITAudit object for testing"""
+    def __init__(self, **kwargs):
+        self.company_id = kwargs.get('company_id', 1)
+
+        # Collaboration flags
+        self.open_to_shared_licensing = kwargs.get('open_to_shared_licensing', False)
+        self.open_to_backup_replication = kwargs.get('open_to_backup_replication', False)
+        self.open_to_teams_federation = kwargs.get('open_to_teams_federation', False)
+        self.open_to_shared_monitoring = kwargs.get('open_to_shared_monitoring', False)
+        self.open_to_collective_purchasing = kwargs.get('open_to_collective_purchasing', False)
+        self.open_to_knowledge_sharing = kwargs.get('open_to_knowledge_sharing', False)
+
+        # Technology flags
+        self.has_azure_ad = kwargs.get('has_azure_ad', False)
+        self.azure_tenant_name = kwargs.get('azure_tenant_name', None)
+        self.has_m365 = kwargs.get('has_m365', False)
+        self.m365_plans = kwargs.get('m365_plans', [])
+        self.has_proxmox_pbs = kwargs.get('has_proxmox_pbs', False)
+        self.monitoring_solution = kwargs.get('monitoring_solution', None)
+
+        # Size info
+        self.employee_count = kwargs.get('employee_count', None)
+
+        # Tech stack
+        self.virtualization_platform = kwargs.get('virtualization_platform', None)
+        self.backup_solution = kwargs.get('backup_solution', None)
+        self.network_firewall_brand = kwargs.get('network_firewall_brand', None)
+        self.erp_system = kwargs.get('erp_system', None)
+
+
+class TestCollaborationMatching(unittest.TestCase):
+    """Test collaboration matching logic without database"""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        # Create service without database (we'll test internal methods)
+        self.service = ITAuditService.__new__(ITAuditService)
+
+    def test_backup_replication_match_both_have_pbs_and_open(self):
+        """Test backup replication match when both companies have PBS and are open"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            has_proxmox_pbs=True,
+            open_to_backup_replication=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            has_proxmox_pbs=True,
+            open_to_backup_replication=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        # Should have exactly 1 match for backup_replication
+        backup_matches = [m for m in matches if m.match_type == 'backup_replication']
+        self.assertEqual(len(backup_matches), 1)
+
+        match = backup_matches[0]
+        self.assertEqual(match.company_a_id, 1)
+        self.assertEqual(match.company_b_id, 2)
+        self.assertEqual(match.match_type, 'backup_replication')
+        self.assertEqual(match.match_score, 90)
+        self.assertIn('Proxmox PBS', match.match_reason)
+
+    def test_backup_replication_no_match_only_one_has_pbs(self):
+        """Test no backup match when only one company has PBS"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            has_proxmox_pbs=True,
+            open_to_backup_replication=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            has_proxmox_pbs=False,  # Company B doesn't have PBS
+            open_to_backup_replication=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        backup_matches = [m for m in matches if m.match_type == 'backup_replication']
+        self.assertEqual(len(backup_matches), 0)
+
+    def test_backup_replication_no_match_not_open(self):
+        """Test no backup match when one company is not open"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            has_proxmox_pbs=True,
+            open_to_backup_replication=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            has_proxmox_pbs=True,
+            open_to_backup_replication=False  # Company B not open
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        backup_matches = [m for m in matches if m.match_type == 'backup_replication']
+        self.assertEqual(len(backup_matches), 0)
+
+    def test_shared_licensing_match_both_have_m365_and_open(self):
+        """Test shared licensing match when both have M365 and are open"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            has_m365=True,
+            m365_plans=['Business Basic', 'E3'],
+            open_to_shared_licensing=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            has_m365=True,
+            m365_plans=['E3', 'E5'],
+            open_to_shared_licensing=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        licensing_matches = [m for m in matches if m.match_type == 'shared_licensing']
+        self.assertEqual(len(licensing_matches), 1)
+
+        match = licensing_matches[0]
+        self.assertEqual(match.match_score, 80)
+        self.assertIn('Microsoft 365', match.match_reason)
+
+    def test_teams_federation_match_both_have_azure_ad_and_open(self):
+        """Test Teams federation match when both have Azure AD and are open"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            has_azure_ad=True,
+            azure_tenant_name='firmaa.onmicrosoft.com',
+            open_to_teams_federation=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            has_azure_ad=True,
+            azure_tenant_name='firmab.onmicrosoft.com',
+            open_to_teams_federation=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        teams_matches = [m for m in matches if m.match_type == 'teams_federation']
+        self.assertEqual(len(teams_matches), 1)
+
+        match = teams_matches[0]
+        self.assertEqual(match.match_score, 85)
+        self.assertIn('Azure AD', match.match_reason)
+
+    def test_shared_monitoring_match_both_use_zabbix(self):
+        """Test shared monitoring match when both use Zabbix"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            monitoring_solution='Zabbix',
+            open_to_shared_monitoring=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            monitoring_solution='Zabbix 6.0',
+            open_to_shared_monitoring=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        monitoring_matches = [m for m in matches if m.match_type == 'shared_monitoring']
+        self.assertEqual(len(monitoring_matches), 1)
+
+        match = monitoring_matches[0]
+        self.assertEqual(match.match_score, 75)
+        self.assertIn('Zabbix', match.match_reason)
+
+    def test_collective_purchasing_match_similar_size(self):
+        """Test collective purchasing match for similar company sizes"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            employee_count='11-50',
+            open_to_collective_purchasing=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            employee_count='51-100',  # Similar size (within 3x ratio)
+            open_to_collective_purchasing=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        purchasing_matches = [m for m in matches if m.match_type == 'collective_purchasing']
+        self.assertEqual(len(purchasing_matches), 1)
+
+        match = purchasing_matches[0]
+        self.assertEqual(match.match_score, 70)
+
+    def test_knowledge_sharing_match_similar_tech_stack(self):
+        """Test knowledge sharing match for similar tech stack"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        audit_a = MockITAudit(
+            company_id=1,
+            virtualization_platform='Proxmox',
+            backup_solution='Veeam',
+            has_azure_ad=True,
+            has_m365=True,
+            open_to_knowledge_sharing=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            virtualization_platform='Proxmox',
+            backup_solution='Veeam',
+            has_azure_ad=True,
+            has_m365=True,
+            open_to_knowledge_sharing=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        knowledge_matches = [m for m in matches if m.match_type == 'knowledge_sharing']
+        self.assertEqual(len(knowledge_matches), 1)
+
+        match = knowledge_matches[0]
+        self.assertEqual(match.match_score, 65)
+        # Should have at least 2 common technologies
+        self.assertIn('common_tech', match.shared_attributes)
+        self.assertGreaterEqual(len(match.shared_attributes['common_tech']), 2)
+
+    def test_multiple_matches_for_same_companies(self):
+        """Test that multiple match types can be found for the same company pair"""
+        company_a = MockCompany(1, "Firma A", "firma-a")
+        company_b = MockCompany(2, "Firma B", "firma-b")
+
+        # Both companies have multiple collaboration opportunities
+        audit_a = MockITAudit(
+            company_id=1,
+            has_proxmox_pbs=True,
+            has_azure_ad=True,
+            has_m365=True,
+            open_to_backup_replication=True,
+            open_to_teams_federation=True,
+            open_to_shared_licensing=True
+        )
+        audit_b = MockITAudit(
+            company_id=2,
+            has_proxmox_pbs=True,
+            has_azure_ad=True,
+            has_m365=True,
+            open_to_backup_replication=True,
+            open_to_teams_federation=True,
+            open_to_shared_licensing=True
+        )
+
+        matches = self.service._check_all_match_types(
+            company_a, audit_a, company_b, audit_b
+        )
+
+        # Should have matches for backup_replication, teams_federation, and shared_licensing
+        match_types = {m.match_type for m in matches}
+        self.assertIn('backup_replication', match_types)
+        self.assertIn('teams_federation', match_types)
+        self.assertIn('shared_licensing', match_types)
+        self.assertEqual(len(matches), 3)
+
+    def test_parse_count_range(self):
+        """Test employee count range parsing"""
+        self.assertEqual(self.service._parse_count_range('1-10'), 5)
+        self.assertEqual(self.service._parse_count_range('11-50'), 30)
+        self.assertEqual(self.service._parse_count_range('51-100'), 75)
+        self.assertEqual(self.service._parse_count_range('101-250'), 175)
+        self.assertEqual(self.service._parse_count_range('251-500'), 375)
+        self.assertEqual(self.service._parse_count_range('500+'), 750)
+        self.assertIsNone(self.service._parse_count_range(None))
+        self.assertIsNone(self.service._parse_count_range('unknown'))
+
+    def test_sizes_similar(self):
+        """Test company size similarity check"""
+        # Within 3x ratio should be similar
+        self.assertTrue(self.service._sizes_similar(30, 75))  # 2.5x
+        self.assertTrue(self.service._sizes_similar(30, 30))  # 1x
+        self.assertTrue(self.service._sizes_similar(30, 50))  # 1.67x
+
+        # More than 3x ratio should not be similar
+        self.assertFalse(self.service._sizes_similar(5, 175))  # 35x
+        self.assertFalse(self.service._sizes_similar(5, 750))  # 150x
+
+        # Zero handling
+        self.assertFalse(self.service._sizes_similar(0, 50))
+        self.assertFalse(self.service._sizes_similar(50, 0))
+
+
+class TestCollaborationMatchDataclass(unittest.TestCase):
+    """Test CollaborationMatch dataclass"""
+
+    def test_collaboration_match_creation(self):
+        """Test creating a CollaborationMatch"""
+        match = CollaborationMatch(
+            company_a_id=1,
+            company_b_id=2,
+            company_a_name="Firma A",
+            company_b_name="Firma B",
+            match_type="backup_replication",
+            match_reason="Obie firmy używają Proxmox PBS",
+            match_score=90,
+            shared_attributes={"pbs": True}
+        )
+
+        self.assertEqual(match.company_a_id, 1)
+        self.assertEqual(match.company_b_id, 2)
+        self.assertEqual(match.match_type, "backup_replication")
+        self.assertEqual(match.match_score, 90)
+        self.assertEqual(match.shared_attributes["pbs"], True)
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)