#!/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)