#!/usr/bin/env python3
"""
Active Directory test environment script.
Tests ActiveDirectoryMCP with LDAP/AD test container.
"""
import os
import sys
import json
import time
import requests
# Add src to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from active_directory_mcp.config.loader import load_config
from active_directory_mcp.core.ldap_manager import LDAPManager
from active_directory_mcp.tools.user import UserTools
from active_directory_mcp.tools.group import GroupTools
def wait_for_ad_ready(max_wait=120):
"""Wait for Samba AD to be ready."""
print("š Waiting for Samba AD to be ready...")
for i in range(max_wait):
try:
# Try to connect to LDAP port
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
result = sock.connect_ex(('192.168.1.100', 389))
sock.close()
if result == 0:
print(f"ā
Samba AD is ready! (took {i+1} seconds)")
time.sleep(5) # Extra wait for full initialization
return True
except Exception:
pass
if i % 10 == 0:
print(f" Still waiting... ({i+1}/{max_wait}s)")
time.sleep(1)
print("ā Timeout waiting for Samba AD")
return False
def test_samba_ad_connection():
"""Test connection to Samba AD."""
try:
# Load Samba AD config
config = load_config('ad-config/ad-config.json')
print(f"ā
Config loaded: {config.active_directory.server}")
# Initialize LDAP manager
ldap_manager = LDAPManager(
config.active_directory,
config.security,
config.performance
)
print("ā
LDAP Manager initialized")
# Test connection
connection_result = ldap_manager.test_connection()
if connection_result.get('connected'):
print(f"ā
Connected to Samba AD: {connection_result.get('server')}")
else:
print(f"ā Connection failed: {connection_result.get('error')}")
return False
# Test user tools
print("\nš„ Testing User Management...")
user_tools = UserTools(ldap_manager)
# List users
users = user_tools.list_users()
if users and len(users) > 0:
user_data = json.loads(users[0].text)
if user_data.get('success'):
user_list = user_data.get('data', [])
print(f"ā
Found {len(user_list)} users:")
for user in user_list[:5]: # Show first 5
print(f" š {user.get('name')} ({user.get('username')})")
else:
print(f"ā User query failed: {user_data.get('error')}")
# Test group tools
print("\nš Testing Group Management...")
group_tools = GroupTools(ldap_manager)
groups = group_tools.list_groups()
if groups and len(groups) > 0:
group_data = json.loads(groups[0].text)
if group_data.get('success'):
group_list = group_data.get('data', [])
print(f"ā
Found {len(group_list)} groups:")
for group in group_list[:5]: # Show first 5
print(f" š„ {group.get('name')} - {group.get('description', 'No description')}")
else:
print(f"ā Group query failed: {group_data.get('error')}")
return True
except Exception as e:
print(f"ā Error: {e}")
return False
def test_http_api():
"""Test HTTP API endpoints."""
print("\nš Testing HTTP API...")
base_url = "http://localhost:8813/activedirectory-mcp"
try:
# Test health endpoint
response = requests.post(
base_url,
json={"method": "health", "params": {}},
timeout=10
)
if response.status_code == 200:
result = response.json()
print(f"ā
HTTP API health check: {result}")
else:
print(f"ā HTTP API failed: {response.status_code}")
return False
# Test list users via HTTP
response = requests.post(
base_url,
json={"method": "list_users", "params": {}},
timeout=10
)
if response.status_code == 200:
result = response.json()
print("ā
HTTP API user listing successful")
else:
print(f"ā HTTP API user listing failed: {response.status_code}")
return True
except Exception as e:
print(f"ā HTTP API error: {e}")
return False
def main():
"""Main test function."""
print("š¢ Samba Active Directory Test with ActiveDirectoryMCP")
print("=" * 60)
print("\nš Test Environment:")
print(" š³ Samba AD Domain Controller: test.local")
print(" š HTTP API: http://localhost:8813/activedirectory-mcp")
print(" š„ļø LDAP Admin: http://localhost:8080")
print(" š¤ Admin User: Administrator / Admin123!")
# Wait for AD to be ready
if not wait_for_ad_ready():
print("\nš” To start Samba AD:")
print(" docker-compose -f docker-samba-ad.yml up -d")
print(" # Wait 2-3 minutes for full initialization")
return
print("\nš Starting tests...")
# Test direct connection
print("\n" + "="*50)
print("š” Direct LDAP Connection Test")
print("="*50)
success = test_samba_ad_connection()
# Test HTTP API
print("\n" + "="*50)
print("š HTTP API Test")
print("="*50)
http_success = test_http_api()
# Summary
print("\n" + "="*50)
print("š Test Summary")
print("="*50)
if success:
print("ā
Direct LDAP connection: SUCCESS")
else:
print("ā Direct LDAP connection: FAILED")
if http_success:
print("ā
HTTP API: SUCCESS")
else:
print("ā HTTP API: FAILED")
if success and http_success:
print("\nš All tests passed! Samba AD integration successful!")
print("\nš Next steps:")
print(" 1. Use HTTP API in Cursor/VS Code")
print(" 2. Explore LDAP Admin: http://localhost:8080")
print(" 3. Test with real AD operations")
else:
print("\nš” Troubleshooting:")
print(" 1. Check Docker containers: docker ps")
print(" 2. Check AD logs: docker logs samba-ad-dc")
print(" 3. Verify network connectivity")
print(" 4. Wait longer for AD initialization")
if __name__ == "__main__":
main()
