test_upload.py

#!/usr/bin/env python3
"""Test script to verify package upload functionality"""

import requests
import json
import os
import tempfile

BASE_URL = "http://localhost:5001"

def test_login():
    """Test login and get token"""
    login_data = {
        "username": "testuser",
        "password": "testpass123"
    }
    
    response = requests.post(f"{BASE_URL}/api/auth/login", json=login_data)
    
    if response.status_code == 200:
        token = response.json()['access_token']
        print("✓ Login successful")
        return token
    else:
        print(f"✗ Login failed: {response.status_code} - {response.text}")
        return None

def test_create_org(token):
    """Test organization creation"""
    headers = {"Authorization": f"Bearer {token}"}
    org_data = {
        "name": "test-org",
        "description": "Test organization for uploads",
        "is_public": True
    }
    
    response = requests.post(f"{BASE_URL}/api/organizations", json=org_data, headers=headers)
    
    if response.status_code in [200, 201, 409]:  # 409 = already exists
        print("✓ Organization ready")
        return True
    else:
        print(f"✗ Organization creation failed: {response.status_code} - {response.text}")
        return False

def test_create_repo(token, package_type):
    """Test repository creation"""
    headers = {"Authorization": f"Bearer {token}"}
    repo_data = {
        "name": f"test-{package_type}-repo",
        "description": f"Test {package_type} repository",
        "package_type": package_type,
        "visibility": "public"
    }
    
    response = requests.post(f"{BASE_URL}/api/organizations/test-org/repositories", json=repo_data, headers=headers)
    
    if response.status_code in [200, 201, 409]:  # 409 = already exists
        print(f"✓ {package_type} repository ready")
        return True
    else:
        print(f"✗ {package_type} repository creation failed: {response.status_code} - {response.text}")
        return False

def test_upload_file(token, package_type):
    """Test file upload"""
    headers = {"Authorization": f"Bearer {token}"}
    
    # Create a test file
    with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
        f.write(f"Test {package_type} package content\nThis is a test file for the {package_type} repository.")
        test_file_path = f.name
    
    try:
        with open(test_file_path, 'rb') as f:
            files = {'file': (f'test-{package_type}-package.txt', f, 'text/plain')}
            data = {
                'name': f'test-{package_type}-package',
                'version': '1.0.0',
                'description': f'Test {package_type} package'
            }
            
            response = requests.post(
                f"{BASE_URL}/api/organizations/test-org/repositories/test-{package_type}-repo/artifacts",
                files=files,
                data=data,
                headers=headers
            )
            
            if response.status_code in [200, 201]:
                print(f"✓ {package_type} file upload successful")
                return True
            else:
                print(f"✗ {package_type} file upload failed: {response.status_code} - {response.text}")
                return False
    finally:
        os.unlink(test_file_path)

def test_list_artifacts(token, package_type):
    """Test listing artifacts"""
    headers = {"Authorization": f"Bearer {token}"}
    
    response = requests.get(
        f"{BASE_URL}/api/organizations/test-org/repositories/test-{package_type}-repo/artifacts",
        headers=headers
    )
    
    if response.status_code == 200:
        artifacts = response.json()
        print(f"✓ {package_type} artifacts listed: {len(artifacts)} found")
        return True
    else:
        print(f"✗ {package_type} artifacts listing failed: {response.status_code} - {response.text}")
        return False

def test_docker_registry():
    """Test Docker Registry API V2"""
    try:
        response = requests.get(f"{BASE_URL}/v2/")
        
        if response.status_code == 200:
            print("✓ Docker Registry API V2 endpoint responding")
            return True
        else:
            print(f"✗ Docker Registry API V2 failed: {response.status_code}")
            return False
    except Exception as e:
        print(f"✗ Docker Registry API V2 error: {e}")
        return False

def main():
    print("Testing Artifact Repository Functionality\n")
    
    # Test basic authentication
    token = test_login()
    if not token:
        print("Cannot proceed without authentication")
        return
    
    # Test organization setup
    if not test_create_org(token):
        print("Cannot proceed without organization")
        return
    
    # Test different package types
    package_types = ['docker', 'npm', 'python', 'maven', 'generic']
    
    for package_type in package_types:
        print(f"\n--- Testing {package_type.upper()} ---")
        
        if test_create_repo(token, package_type):
            if test_upload_file(token, package_type):
                test_list_artifacts(token, package_type)
    
    # Test Docker Registry API
    print(f"\n--- Testing Docker Registry API ---")
    test_docker_registry()
    
    print(f"\n🎉 Testing complete!")
    print(f"You can now:")
    print(f"1. View artifacts in the web UI")
    print(f"2. Try Docker commands with your registry")
    print(f"3. Upload more files through the web interface")

if __name__ == "__main__":
    main()