nordabiz/scripts/test_collaboration_matching.py

#!/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:
    # Set DATABASE_URL environment variable with proper credentials, then run:
    DATABASE_URL=postgresql://nordabiz_app:YOUR_PASSWORD@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: Maciej Pienczyn, InPi sp. z o.o.
Created: 2026-01-09
"""

import os
import sys
from datetime import datetime

# Set database URL for DEV environment
# WARNING: The fallback DATABASE_URL uses a placeholder password.
# Production credentials MUST be set via the DATABASE_URL environment variable.
# NEVER commit real credentials to version control (CWE-798).
if 'DATABASE_URL' not in os.environ:
    os.environ['DATABASE_URL'] = 'postgresql://nordabiz_app:CHANGE_ME@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)