Story 1.6 finished

main
PeninsulaInd 2025-10-18 11:35:47 -05:00
parent b6e495e9fe
commit 4cada9df42
14 changed files with 1932 additions and 13 deletions

View File

View File

@ -55,15 +55,18 @@ After this story is complete, copy .env.example to .env and populate it with the
- The commands can only be run by an authenticated "Admin".
- Appropriate feedback is provided to the console upon success or failure of the commands.
### Story 1.6: FQDN Bucket Mapping Management
**As an Admin**, I want a database table and CLI commands to map cloud bucket names to custom FQDNs, so that the system can generate correct public URLs for interlinking.
### Story 1.6: Deployment Infrastructure Management
**As an Admin**, I want a database model and CLI commands to manage bunny.net deployment infrastructure for static sites, so that the system can provision new sites, attach domains to existing storage, and track all necessary credentials and IDs for future operations.
**Acceptance Criteria**
- A new database table, fqdn_mappings, is created with columns for bucket_name, provider, and fqdn.
- A CLI command exists for an Admin to add a new mapping (e.g., add-fqdn-mapping --bucket my-bunny-bucket --provider bunny --fqdn www.mycustomdomain.com).
- A CLI command exists for an Admin to remove a mapping.
- A CLI command exists for an Admin to list all current mappings.
- The commands are protected and can only be run by an authenticated "Admin".
- A `SiteDeployment` database table stores all bunny.net resource details (Storage Zone ID/password/region, Pull Zone ID/hostname, custom domain).
- CLI command `provision-site` creates new Storage Zone + Pull Zone + custom hostname on bunny.net and stores all credentials.
- CLI command `attach-domain` creates a Pull Zone linked to an existing Storage Zone and adds a custom hostname.
- CLI commands `list-sites`, `get-site`, and `remove-site` manage deployment records.
- Provisioning commands display DNS configuration instructions for manual CNAME setup by the user.
- All commands are Admin-only and integrate with the bunny.net Account API.
**See**: [Story 1.6 Detailed Requirements](../stories/story-1.6-deployment-infrastructure.md)
### Story 1.7: CI/CD Pipeline Setup
**As a developer**, I want a basic CI/CD pipeline configured for the project, so that code changes are automatically tested and can be deployed consistently.

View File

@ -319,7 +319,7 @@ Test Results:
## Next Steps
This CLI user management foundation is now ready for:
- **Story 1.6**: FQDN Bucket Mapping Management (CLI commands for domain mappings)
- **Story 1.6**: Deployment Infrastructure Management (CLI commands for bunny.net provisioning)
- **Story 1.7**: CI/CD Pipeline Setup (automated testing in pipeline)
- **Epic 2**: Content Generation features will use user authentication
- Integration with future admin dashboard/web interface

View File

@ -0,0 +1,142 @@
# Story 1.6: Deployment Infrastructure Management
## Story
**As an Admin**, I want a database model and CLI commands to manage bunny.net deployment infrastructure for static sites, so that the system can provision new sites, attach domains to existing storage, and track all necessary credentials and IDs for future operations.
## Context
bunny.net requires a sequential provisioning workflow using two separate APIs (Account API for infrastructure, Storage API for files). Each site deployment requires tracking multiple related resources (Storage Zones and Pull Zones) with their corresponding IDs, credentials, and hostnames for future management operations like file uploads, cache purges, and deletion.
## Acceptance Criteria
### Database Model
- A `SiteDeployment` table exists with the following fields:
- `id` (integer, primary key)
- `site_name` (string, required) - User-friendly name for the site
- `custom_hostname` (string, required, unique) - The FQDN (e.g., www.yourdomain.com)
- `storage_zone_id` (integer, required) - bunny.net Storage Zone ID
- `storage_zone_name` (string, required) - Storage Zone name
- `storage_zone_password` (string, required) - Storage Zone API password
- `storage_zone_region` (string, required) - Storage region code (e.g., "DE", "NY", "LA")
- `pull_zone_id` (integer, required) - bunny.net Pull Zone ID
- `pull_zone_bcdn_hostname` (string, required) - Default b-cdn.net hostname
- `created_at` (timestamp)
- `updated_at` (timestamp)
### CLI Commands
#### Provision New Site
- Command: `provision-site --name <site_name> --domain <custom_hostname> --storage-name <storage_zone_name> --region <region>`
- Creates a new Storage Zone on bunny.net
- Creates a new Pull Zone linked to the Storage Zone
- Adds the custom hostname to the Pull Zone
- Stores all response data in the database
- Displays DNS configuration instructions for manual setup by the user
- Admin-only access
#### Attach Domain to Existing Storage
- Command: `attach-domain --name <site_name> --domain <custom_hostname> --storage-name <existing_storage_zone_name>`
- Retrieves existing Storage Zone ID by name
- Creates a new Pull Zone linked to the existing Storage Zone
- Adds the custom hostname to the Pull Zone
- Stores all response data in the database
- Displays DNS configuration instructions for manual setup by the user
- Admin-only access
#### List Sites
- Command: `list-sites`
- Displays all site deployments with key details (site name, custom hostname, storage zone name, pull zone b-cdn hostname)
- Admin-only access
#### Get Site Details
- Command: `get-site --domain <custom_hostname>`
- Displays complete deployment details for a specific site
- Admin-only access
#### Remove Site
- Command: `remove-site --domain <custom_hostname>`
- Removes the site deployment record from the database
- Does NOT delete resources from bunny.net (manual cleanup required)
- Requires confirmation prompt
- Admin-only access
### API Integration Requirements
#### Configuration
- `BUNNY_ACCOUNT_API_KEY` must be present in environment variables
- Base URL: `https://api.bunny.net`
- All requests require headers:
- `AccessKey`: Account API Key
- `Content-Type`: `application/json`
#### Workflow A: New Site Provisioning
1. **Create Storage Zone**
- Endpoint: `POST /storagezone`
- Request: `{"Name": "<storage_zone_name>", "Region": "<region>"}`
- Capture: `Id`, `Password`
2. **Create Pull Zone**
- Endpoint: `POST /pullzone`
- Request: `{"Name": "<pull_zone_name>", "OriginType": 2, "StorageZoneId": <storage_zone_id>}`
- Capture: `Id`, `Hostname`
3. **Add Custom Hostname**
- Endpoint: `POST /pullzone/{pull_zone_id}/addHostname`
- Request: `{"Hostname": "<custom_hostname>"}`
- Verify: 200 OK response
#### Workflow B: Attach to Existing Storage
1. **Find Existing Storage Zone**
- Endpoint: `GET /storagezone`
- Parse response to find zone by name
- Capture: `Id`, `Password`, `Region`
2. **Create Pull Zone** (same as Workflow A, step 2)
3. **Add Custom Hostname** (same as Workflow A, step 3)
### Error Handling
- Handle API authentication failures (invalid API key)
- Handle resource name conflicts (Storage/Pull Zone names must be unique)
- Handle missing Storage Zones (Workflow B)
- Handle network errors with appropriate retry logic
- Validate region codes before API calls
- Validate FQDN format before provisioning
### Output Requirements
- After successful provisioning, display DNS configuration instructions for the user to manually configure with their domain registrar:
```
Site provisioned successfully!
MANUAL DNS CONFIGURATION REQUIRED:
You must create the following CNAME record with your domain registrar:
Type: CNAME
Host: <subdomain>
Value: <pull_zone_bcdn_hostname>
Example:
Type: CNAME
Host: www
Value: my-site-cdn.b-cdn.net
Note: DNS propagation may take up to 48 hours.
```
## Technical Notes
- Storage Zone names and Pull Zone names must be globally unique across bunny.net
- The Storage Zone password is only returned during creation; it cannot be retrieved later
- OriginType must always be 2 for Storage Zone-backed Pull Zones
- **DNS Configuration**: This system does NOT automate DNS record creation. The user must manually create CNAME records with their domain registrar. The CLI only provides instructions.
- DNS propagation may take up to 48 hours after the user creates the CNAME record
## Dependencies
- Story 1.5 (CLI infrastructure and admin authentication)
- Story 1.2 (Database setup and ORM)
- bunny.net Account API Key in environment configuration
## Future Considerations
- Story 4.x will use these stored credentials for file upload operations
- Story 4.x will use these IDs for cache purge operations
- Deletion of bunny.net resources (Storage/Pull Zones) will require separate implementation
- Multi-region replication configuration may be added later

View File

@ -16,6 +16,7 @@ AZURE_STORAGE_ACCOUNT_NAME=your_azure_account_name_here
AZURE_STORAGE_ACCOUNT_KEY=your_azure_account_key_here
# Bunny.net Configuration
BUNNY_ACCOUNT_API_KEY=your_bunny_account_api_key_here
BUNNY_API_KEY=your_bunny_api_key_here
BUNNY_STORAGE_ZONE=your_bunny_zone_here

View File

@ -4,11 +4,17 @@ CLI command definitions using Click
import click
from typing import Optional
from src.core.config import get_config
from src.core.config import get_config, get_bunny_account_api_key
from src.auth.service import AuthService
from src.database.session import db_manager
from src.database.repositories import UserRepository
from src.database.repositories import UserRepository, SiteDeploymentRepository
from src.database.models import User
from src.deployment.bunnynet import (
BunnyNetClient,
BunnyNetAPIError,
BunnyNetAuthError,
BunnyNetResourceConflictError
)
def authenticate_admin(username: str, password: str) -> Optional[User]:
@ -239,5 +245,361 @@ def list_users(admin_user: Optional[str], admin_password: Optional[str]):
raise click.Abort()
@app.command("provision-site")
@click.option("--name", prompt=True, help="Site name")
@click.option("--domain", prompt=True, help="Custom domain (FQDN, e.g., www.example.com)")
@click.option("--storage-name", prompt=True, help="Storage Zone name (must be globally unique)")
@click.option("--region", prompt=True, type=click.Choice(["DE", "NY", "LA", "SG", "SYD"]),
help="Storage region")
@click.option("--admin-user", help="Admin username for authentication")
@click.option("--admin-password", help="Admin password for authentication")
def provision_site(name: str, domain: str, storage_name: str, region: str,
admin_user: Optional[str], admin_password: Optional[str]):
"""Provision a new site with Storage Zone and Pull Zone (requires admin)"""
try:
# Authenticate admin
if not admin_user or not admin_password:
admin_user, admin_password = prompt_admin_credentials()
admin = authenticate_admin(admin_user, admin_password)
if not admin:
click.echo("Error: Authentication failed or insufficient permissions", err=True)
raise click.Abort()
# Get bunny.net API key
try:
api_key = get_bunny_account_api_key()
except ValueError as e:
click.echo(f"Error: {e}", err=True)
click.echo("Please set BUNNY_ACCOUNT_API_KEY in your .env file", err=True)
raise click.Abort()
click.echo(f"\nProvisioning site '{name}' with domain '{domain}'...")
# Initialize bunny.net client
client = BunnyNetClient(api_key)
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
# Check if domain already exists
if deployment_repo.exists(domain):
click.echo(f"Error: Site with domain '{domain}' already exists", err=True)
raise click.Abort()
# Step 1: Create Storage Zone
click.echo(f"Step 1/3: Creating Storage Zone '{storage_name}' in region {region}...")
storage_result = client.create_storage_zone(storage_name, region)
click.echo(f" Storage Zone created: ID={storage_result.id}")
# Step 2: Create Pull Zone
pull_zone_name = f"{storage_name}-cdn"
click.echo(f"Step 2/3: Creating Pull Zone '{pull_zone_name}'...")
pull_result = client.create_pull_zone(pull_zone_name, storage_result.id)
click.echo(f" Pull Zone created: ID={pull_result.id}, Hostname={pull_result.hostname}")
# Step 3: Add Custom Hostname
click.echo(f"Step 3/3: Adding custom hostname '{domain}'...")
client.add_custom_hostname(pull_result.id, domain)
click.echo(f" Custom hostname added successfully")
# Save to database
deployment = deployment_repo.create(
site_name=name,
custom_hostname=domain,
storage_zone_id=storage_result.id,
storage_zone_name=storage_result.name,
storage_zone_password=storage_result.password,
storage_zone_region=storage_result.region,
pull_zone_id=pull_result.id,
pull_zone_bcdn_hostname=pull_result.hostname
)
click.echo("\n" + "=" * 70)
click.echo("Site provisioned successfully!")
click.echo("=" * 70)
click.echo("\nMANUAL DNS CONFIGURATION REQUIRED:")
click.echo("You must create the following CNAME record with your domain registrar:\n")
click.echo(f" Type: CNAME")
subdomain = domain.split('.')[0] if '.' in domain else '@'
click.echo(f" Host: {subdomain}")
click.echo(f" Value: {pull_result.hostname}")
click.echo("\nExample DNS configuration:")
click.echo(f" Type: CNAME")
click.echo(f" Host: {subdomain}")
click.echo(f" Value: {pull_result.hostname}")
click.echo("\nNote: DNS propagation may take up to 48 hours.")
click.echo("=" * 70)
except BunnyNetAuthError as e:
click.echo(f"Error: Authentication failed - {e}", err=True)
click.echo("Please check your BUNNY_ACCOUNT_API_KEY", err=True)
raise click.Abort()
except BunnyNetResourceConflictError as e:
click.echo(f"Error: Resource conflict - {e}", err=True)
click.echo("Storage Zone or Pull Zone name already exists. Try a different name.", err=True)
raise click.Abort()
except BunnyNetAPIError as e:
click.echo(f"Error: bunny.net API error - {e}", err=True)
raise click.Abort()
finally:
session.close()
except Exception as e:
click.echo(f"Error provisioning site: {e}", err=True)
raise click.Abort()
@app.command("attach-domain")
@click.option("--name", prompt=True, help="Site name")
@click.option("--domain", prompt=True, help="Custom domain (FQDN, e.g., www.example.com)")
@click.option("--storage-name", prompt=True, help="Existing Storage Zone name")
@click.option("--admin-user", help="Admin username for authentication")
@click.option("--admin-password", help="Admin password for authentication")
def attach_domain(name: str, domain: str, storage_name: str,
admin_user: Optional[str], admin_password: Optional[str]):
"""Attach a domain to an existing Storage Zone (requires admin)"""
try:
# Authenticate admin
if not admin_user or not admin_password:
admin_user, admin_password = prompt_admin_credentials()
admin = authenticate_admin(admin_user, admin_password)
if not admin:
click.echo("Error: Authentication failed or insufficient permissions", err=True)
raise click.Abort()
# Get bunny.net API key
try:
api_key = get_bunny_account_api_key()
except ValueError as e:
click.echo(f"Error: {e}", err=True)
click.echo("Please set BUNNY_ACCOUNT_API_KEY in your .env file", err=True)
raise click.Abort()
click.echo(f"\nAttaching domain '{domain}' to existing Storage Zone '{storage_name}'...")
# Initialize bunny.net client
client = BunnyNetClient(api_key)
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
# Check if domain already exists
if deployment_repo.exists(domain):
click.echo(f"Error: Site with domain '{domain}' already exists", err=True)
raise click.Abort()
# Step 1: Find existing Storage Zone
click.echo(f"Step 1/3: Finding Storage Zone '{storage_name}'...")
storage_result = client.find_storage_zone_by_name(storage_name)
if not storage_result:
click.echo(f"Error: Storage Zone '{storage_name}' not found", err=True)
raise click.Abort()
click.echo(f" Storage Zone found: ID={storage_result.id}")
# Step 2: Create Pull Zone
pull_zone_name = f"{storage_name}-{domain.replace('.', '-')}"
click.echo(f"Step 2/3: Creating Pull Zone '{pull_zone_name}'...")
pull_result = client.create_pull_zone(pull_zone_name, storage_result.id)
click.echo(f" Pull Zone created: ID={pull_result.id}, Hostname={pull_result.hostname}")
# Step 3: Add Custom Hostname
click.echo(f"Step 3/3: Adding custom hostname '{domain}'...")
client.add_custom_hostname(pull_result.id, domain)
click.echo(f" Custom hostname added successfully")
# Save to database
deployment = deployment_repo.create(
site_name=name,
custom_hostname=domain,
storage_zone_id=storage_result.id,
storage_zone_name=storage_result.name,
storage_zone_password=storage_result.password,
storage_zone_region=storage_result.region,
pull_zone_id=pull_result.id,
pull_zone_bcdn_hostname=pull_result.hostname
)
click.echo("\n" + "=" * 70)
click.echo("Domain attached successfully!")
click.echo("=" * 70)
click.echo("\nMANUAL DNS CONFIGURATION REQUIRED:")
click.echo("You must create the following CNAME record with your domain registrar:\n")
click.echo(f" Type: CNAME")
subdomain = domain.split('.')[0] if '.' in domain else '@'
click.echo(f" Host: {subdomain}")
click.echo(f" Value: {pull_result.hostname}")
click.echo("\nExample DNS configuration:")
click.echo(f" Type: CNAME")
click.echo(f" Host: {subdomain}")
click.echo(f" Value: {pull_result.hostname}")
click.echo("\nNote: DNS propagation may take up to 48 hours.")
click.echo("=" * 70)
except BunnyNetAuthError as e:
click.echo(f"Error: Authentication failed - {e}", err=True)
click.echo("Please check your BUNNY_ACCOUNT_API_KEY", err=True)
raise click.Abort()
except BunnyNetResourceConflictError as e:
click.echo(f"Error: Resource conflict - {e}", err=True)
click.echo("Pull Zone name already exists. Try a different domain.", err=True)
raise click.Abort()
except BunnyNetAPIError as e:
click.echo(f"Error: bunny.net API error - {e}", err=True)
raise click.Abort()
finally:
session.close()
except Exception as e:
click.echo(f"Error attaching domain: {e}", err=True)
raise click.Abort()
@app.command("list-sites")
@click.option("--admin-user", help="Admin username for authentication")
@click.option("--admin-password", help="Admin password for authentication")
def list_sites(admin_user: Optional[str], admin_password: Optional[str]):
"""List all site deployments (requires admin)"""
try:
# Authenticate admin
if not admin_user or not admin_password:
admin_user, admin_password = prompt_admin_credentials()
admin = authenticate_admin(admin_user, admin_password)
if not admin:
click.echo("Error: Authentication failed or insufficient permissions", err=True)
raise click.Abort()
# List all sites
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
sites = deployment_repo.get_all()
if not sites:
click.echo("No site deployments found")
return
click.echo(f"\nTotal sites: {len(sites)}")
click.echo("-" * 100)
click.echo(f"{'ID':<5} {'Site Name':<25} {'Custom Domain':<30} {'Storage Zone':<20} {'Region':<8}")
click.echo("-" * 100)
for site in sites:
click.echo(f"{site.id:<5} {site.site_name:<25} {site.custom_hostname:<30} "
f"{site.storage_zone_name:<20} {site.storage_zone_region:<8}")
click.echo("-" * 100)
finally:
session.close()
except Exception as e:
click.echo(f"Error listing sites: {e}", err=True)
raise click.Abort()
@app.command("get-site")
@click.option("--domain", prompt=True, help="Custom domain to lookup")
@click.option("--admin-user", help="Admin username for authentication")
@click.option("--admin-password", help="Admin password for authentication")
def get_site(domain: str, admin_user: Optional[str], admin_password: Optional[str]):
"""Get detailed information about a site deployment (requires admin)"""
try:
# Authenticate admin
if not admin_user or not admin_password:
admin_user, admin_password = prompt_admin_credentials()
admin = authenticate_admin(admin_user, admin_password)
if not admin:
click.echo("Error: Authentication failed or insufficient permissions", err=True)
raise click.Abort()
# Get site details
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
site = deployment_repo.get_by_hostname(domain)
if not site:
click.echo(f"Error: Site with domain '{domain}' not found", err=True)
raise click.Abort()
click.echo("\n" + "=" * 70)
click.echo("Site Deployment Details")
click.echo("=" * 70)
click.echo(f"ID: {site.id}")
click.echo(f"Site Name: {site.site_name}")
click.echo(f"Custom Domain: {site.custom_hostname}")
click.echo(f"\nStorage Zone:")
click.echo(f" ID: {site.storage_zone_id}")
click.echo(f" Name: {site.storage_zone_name}")
click.echo(f" Region: {site.storage_zone_region}")
click.echo(f" Password: {site.storage_zone_password}")
click.echo(f"\nPull Zone:")
click.echo(f" ID: {site.pull_zone_id}")
click.echo(f" b-cdn Hostname: {site.pull_zone_bcdn_hostname}")
click.echo(f"\nTimestamps:")
click.echo(f" Created: {site.created_at.strftime('%Y-%m-%d %H:%M:%S')}")
click.echo(f" Updated: {site.updated_at.strftime('%Y-%m-%d %H:%M:%S')}")
click.echo("=" * 70)
finally:
session.close()
except Exception as e:
click.echo(f"Error getting site details: {e}", err=True)
raise click.Abort()
@app.command("remove-site")
@click.option("--domain", prompt=True, help="Custom domain to remove")
@click.option("--admin-user", help="Admin username for authentication")
@click.option("--admin-password", help="Admin password for authentication")
@click.confirmation_option(prompt="Are you sure you want to remove this site deployment record?")
def remove_site(domain: str, admin_user: Optional[str], admin_password: Optional[str]):
"""Remove a site deployment record (requires admin)"""
try:
# Authenticate admin
if not admin_user or not admin_password:
admin_user, admin_password = prompt_admin_credentials()
admin = authenticate_admin(admin_user, admin_password)
if not admin:
click.echo("Error: Authentication failed or insufficient permissions", err=True)
raise click.Abort()
# Remove site
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
# Check if site exists
site = deployment_repo.get_by_hostname(domain)
if not site:
click.echo(f"Error: Site with domain '{domain}' not found", err=True)
raise click.Abort()
# Delete the site
success = deployment_repo.delete(site.id)
if success:
click.echo(f"Success: Site deployment record for '{domain}' has been removed")
click.echo("\nNote: This does NOT delete resources from bunny.net.")
click.echo("You must manually delete the Storage Zone and Pull Zone if needed.")
else:
click.echo(f"Error: Failed to remove site '{domain}'", err=True)
raise click.Abort()
finally:
session.close()
except Exception as e:
click.echo(f"Error removing site: {e}", err=True)
raise click.Abort()
if __name__ == "__main__":
app()

View File

@ -170,3 +170,11 @@ def get_ai_api_key() -> str:
if not api_key:
raise ValueError("AI_API_KEY environment variable is required")
return api_key
def get_bunny_account_api_key() -> str:
"""Get the bunny.net Account API key from environment variables"""
api_key = os.getenv("BUNNY_ACCOUNT_API_KEY")
if not api_key:
raise ValueError("BUNNY_ACCOUNT_API_KEY environment variable is required")
return api_key

View File

@ -4,7 +4,7 @@ Abstract repository interfaces for data access layer
from abc import ABC, abstractmethod
from typing import Optional, List
from src.database.models import User
from src.database.models import User, SiteDeployment
class IUserRepository(ABC):
@ -44,3 +44,47 @@ class IUserRepository(ABC):
def exists(self, username: str) -> bool:
"""Check if a user exists by username"""
pass
class ISiteDeploymentRepository(ABC):
"""Interface for SiteDeployment data access"""
@abstractmethod
def create(
self,
site_name: str,
custom_hostname: str,
storage_zone_id: int,
storage_zone_name: str,
storage_zone_password: str,
storage_zone_region: str,
pull_zone_id: int,
pull_zone_bcdn_hostname: str
) -> SiteDeployment:
"""Create a new site deployment"""
pass
@abstractmethod
def get_by_id(self, deployment_id: int) -> Optional[SiteDeployment]:
"""Get a site deployment by ID"""
pass
@abstractmethod
def get_by_hostname(self, custom_hostname: str) -> Optional[SiteDeployment]:
"""Get a site deployment by custom hostname"""
pass
@abstractmethod
def get_all(self) -> List[SiteDeployment]:
"""Get all site deployments"""
pass
@abstractmethod
def delete(self, deployment_id: int) -> bool:
"""Delete a site deployment by ID"""
pass
@abstractmethod
def exists(self, custom_hostname: str) -> bool:
"""Check if a site deployment exists by hostname"""
pass

View File

@ -35,3 +35,28 @@ class User(Base):
def is_admin(self) -> bool:
"""Check if user has admin role"""
return self.role == "Admin"
class SiteDeployment(Base):
"""Site deployment model for bunny.net infrastructure tracking"""
__tablename__ = "site_deployments"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
site_name: Mapped[str] = mapped_column(String(255), nullable=False)
custom_hostname: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
storage_zone_id: Mapped[int] = mapped_column(Integer, nullable=False)
storage_zone_name: Mapped[str] = mapped_column(String(255), nullable=False)
storage_zone_password: Mapped[str] = mapped_column(String(255), nullable=False)
storage_zone_region: Mapped[str] = mapped_column(String(10), nullable=False)
pull_zone_id: Mapped[int] = mapped_column(Integer, nullable=False)
pull_zone_bcdn_hostname: Mapped[str] = mapped_column(String(255), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
updated_at: Mapped[datetime] = mapped_column(
DateTime,
default=datetime.utcnow,
onupdate=datetime.utcnow,
nullable=False
)
def __repr__(self) -> str:
return f"<SiteDeployment(id={self.id}, site_name='{self.site_name}', custom_hostname='{self.custom_hostname}')>"

View File

@ -5,8 +5,8 @@ Concrete repository implementations
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from src.database.interfaces import IUserRepository
from src.database.models import User
from src.database.interfaces import IUserRepository, ISiteDeploymentRepository
from src.database.models import User, SiteDeployment
class UserRepository(IUserRepository):
@ -124,3 +124,122 @@ class UserRepository(IUserRepository):
True if user exists, False otherwise
"""
return self.session.query(User).filter(User.username == username).first() is not None
class SiteDeploymentRepository(ISiteDeploymentRepository):
"""Repository implementation for SiteDeployment data access"""
def __init__(self, session: Session):
self.session = session
def create(
self,
site_name: str,
custom_hostname: str,
storage_zone_id: int,
storage_zone_name: str,
storage_zone_password: str,
storage_zone_region: str,
pull_zone_id: int,
pull_zone_bcdn_hostname: str
) -> SiteDeployment:
"""
Create a new site deployment
Args:
site_name: User-friendly name for the site
custom_hostname: The FQDN (e.g., www.yourdomain.com)
storage_zone_id: bunny.net Storage Zone ID
storage_zone_name: Storage Zone name
storage_zone_password: Storage Zone API password
storage_zone_region: Storage region code (e.g., "DE", "NY", "LA")
pull_zone_id: bunny.net Pull Zone ID
pull_zone_bcdn_hostname: Default b-cdn.net hostname
Returns:
The created SiteDeployment object
Raises:
ValueError: If custom_hostname already exists
"""
deployment = SiteDeployment(
site_name=site_name,
custom_hostname=custom_hostname,
storage_zone_id=storage_zone_id,
storage_zone_name=storage_zone_name,
storage_zone_password=storage_zone_password,
storage_zone_region=storage_zone_region,
pull_zone_id=pull_zone_id,
pull_zone_bcdn_hostname=pull_zone_bcdn_hostname
)
try:
self.session.add(deployment)
self.session.commit()
self.session.refresh(deployment)
return deployment
except IntegrityError:
self.session.rollback()
raise ValueError(f"Site deployment with hostname '{custom_hostname}' already exists")
def get_by_id(self, deployment_id: int) -> Optional[SiteDeployment]:
"""
Get a site deployment by ID
Args:
deployment_id: The deployment ID to search for
Returns:
SiteDeployment object if found, None otherwise
"""
return self.session.query(SiteDeployment).filter(SiteDeployment.id == deployment_id).first()
def get_by_hostname(self, custom_hostname: str) -> Optional[SiteDeployment]:
"""
Get a site deployment by custom hostname
Args:
custom_hostname: The hostname to search for
Returns:
SiteDeployment object if found, None otherwise
"""
return self.session.query(SiteDeployment).filter(SiteDeployment.custom_hostname == custom_hostname).first()
def get_all(self) -> List[SiteDeployment]:
"""
Get all site deployments
Returns:
List of all SiteDeployment objects
"""
return self.session.query(SiteDeployment).all()
def delete(self, deployment_id: int) -> bool:
"""
Delete a site deployment by ID
Args:
deployment_id: The ID of the deployment to delete
Returns:
True if deleted, False if deployment not found
"""
deployment = self.get_by_id(deployment_id)
if deployment:
self.session.delete(deployment)
self.session.commit()
return True
return False
def exists(self, custom_hostname: str) -> bool:
"""
Check if a site deployment exists by hostname
Args:
custom_hostname: The hostname to check
Returns:
True if deployment exists, False otherwise
"""
return self.session.query(SiteDeployment).filter(SiteDeployment.custom_hostname == custom_hostname).first() is not None

View File

@ -0,0 +1,245 @@
"""
bunny.net API client for managing Storage Zones and Pull Zones
"""
import requests
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
@dataclass
class StorageZoneResult:
"""Result from Storage Zone API call"""
id: int
name: str
password: str
region: str
@dataclass
class PullZoneResult:
"""Result from Pull Zone API call"""
id: int
name: str
hostname: str
class BunnyNetAPIError(Exception):
"""Base exception for bunny.net API errors"""
pass
class BunnyNetAuthError(BunnyNetAPIError):
"""Authentication error with bunny.net API"""
pass
class BunnyNetResourceConflictError(BunnyNetAPIError):
"""Resource name conflict error"""
pass
class BunnyNetClient:
"""Client for interacting with bunny.net Account API"""
BASE_URL = "https://api.bunny.net"
def __init__(self, api_key: str):
"""
Initialize bunny.net API client
Args:
api_key: The bunny.net Account API Key
"""
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"AccessKey": api_key,
"Content-Type": "application/json"
})
def _make_request(
self,
method: str,
endpoint: str,
json_data: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Make an HTTP request to bunny.net API
Args:
method: HTTP method (GET, POST, etc.)
endpoint: API endpoint path
json_data: JSON payload for POST/PUT requests
params: Query parameters
Returns:
API response as dict
Raises:
BunnyNetAuthError: If authentication fails
BunnyNetResourceConflictError: If resource name conflicts
BunnyNetAPIError: For other API errors
"""
url = f"{self.BASE_URL}{endpoint}"
try:
response = self.session.request(
method=method,
url=url,
json=json_data,
params=params,
timeout=30
)
if response.status_code == 401:
raise BunnyNetAuthError("Invalid API key or authentication failed")
if response.status_code == 409:
raise BunnyNetResourceConflictError("Resource name already exists")
if response.status_code >= 400:
error_msg = f"API request failed with status {response.status_code}"
try:
error_data = response.json()
if "Message" in error_data:
error_msg += f": {error_data['Message']}"
except:
error_msg += f": {response.text}"
raise BunnyNetAPIError(error_msg)
if response.status_code == 204 or not response.content:
return {}
return response.json()
except requests.exceptions.Timeout:
raise BunnyNetAPIError("Request timeout")
except requests.exceptions.ConnectionError:
raise BunnyNetAPIError("Connection error")
except requests.exceptions.RequestException as e:
raise BunnyNetAPIError(f"Request failed: {str(e)}")
def create_storage_zone(self, name: str, region: str) -> StorageZoneResult:
"""
Create a new Storage Zone
Args:
name: Unique name for the Storage Zone
region: Storage region code (e.g., "DE", "NY", "LA")
Returns:
StorageZoneResult with ID, name, password, and region
Raises:
BunnyNetResourceConflictError: If name already exists
BunnyNetAPIError: For other API errors
"""
data = {
"Name": name,
"Region": region
}
response = self._make_request("POST", "/storagezone", json_data=data)
return StorageZoneResult(
id=response["Id"],
name=response["Name"],
password=response["Password"],
region=response["Region"]
)
def get_storage_zones(self) -> List[Dict[str, Any]]:
"""
Get all Storage Zones
Returns:
List of Storage Zone dictionaries
Raises:
BunnyNetAPIError: For API errors
"""
response = self._make_request("GET", "/storagezone")
return response if isinstance(response, list) else []
def find_storage_zone_by_name(self, name: str) -> Optional[StorageZoneResult]:
"""
Find a Storage Zone by name
Args:
name: The Storage Zone name to search for
Returns:
StorageZoneResult if found, None otherwise
Raises:
BunnyNetAPIError: For API errors
"""
zones = self.get_storage_zones()
for zone in zones:
if zone.get("Name") == name:
return StorageZoneResult(
id=zone["Id"],
name=zone["Name"],
password=zone.get("Password", ""),
region=zone.get("Region", "")
)
return None
def create_pull_zone(
self,
name: str,
storage_zone_id: int
) -> PullZoneResult:
"""
Create a new Pull Zone linked to a Storage Zone
Args:
name: Unique name for the Pull Zone
storage_zone_id: ID of the Storage Zone to link to
Returns:
PullZoneResult with ID, name, and hostname
Raises:
BunnyNetResourceConflictError: If name already exists
BunnyNetAPIError: For other API errors
"""
data = {
"Name": name,
"OriginType": 2,
"StorageZoneId": storage_zone_id
}
response = self._make_request("POST", "/pullzone", json_data=data)
return PullZoneResult(
id=response["Id"],
name=response["Name"],
hostname=response["Hostnames"][0]["Value"] if response.get("Hostnames") else f"{name}.b-cdn.net"
)
def add_custom_hostname(self, pull_zone_id: int, hostname: str) -> bool:
"""
Add a custom hostname to a Pull Zone
Args:
pull_zone_id: The Pull Zone ID
hostname: The custom FQDN (e.g., www.yourdomain.com)
Returns:
True if successful
Raises:
BunnyNetAPIError: For API errors
"""
data = {
"Hostname": hostname
}
self._make_request("POST", f"/pullzone/{pull_zone_id}/addHostname", json_data=data)
return True

View File

@ -0,0 +1,356 @@
"""
Integration tests for deployment workflow
"""
import pytest
from click.testing import CliRunner
from unittest.mock import patch, Mock
from src.cli.commands import app
from src.database.session import db_manager
from src.database.repositories import UserRepository, SiteDeploymentRepository
from src.auth.service import AuthService
from src.deployment.bunnynet import StorageZoneResult, PullZoneResult
@pytest.fixture
def cli_runner():
"""Fixture to create a Click CLI test runner"""
return CliRunner()
@pytest.fixture
def setup_test_admin():
"""Create a test admin user in the database"""
session = db_manager.get_session()
try:
user_repo = UserRepository(session)
auth_service = AuthService(user_repo)
try:
user = auth_service.create_user_with_hashed_password(
username="test_admin",
password="test_password",
role="Admin"
)
yield user
finally:
if user_repo.get_by_username("test_admin"):
test_user = user_repo.get_by_username("test_admin")
if test_user:
user_repo.delete(test_user.id)
finally:
session.close()
@pytest.fixture
def cleanup_test_sites():
"""Clean up any test site deployments after test"""
yield
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
sites = deployment_repo.get_all()
for site in sites:
if "test" in site.site_name.lower() or "test" in site.custom_hostname.lower():
deployment_repo.delete(site.id)
finally:
session.close()
class TestProvisionSiteIntegration:
"""Integration tests for provision-site workflow"""
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
def test_full_provisioning_workflow(
self, mock_client_class, mock_get_key, cli_runner, setup_test_admin, cleanup_test_sites
):
"""Test complete site provisioning workflow with database"""
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.create_storage_zone.return_value = StorageZoneResult(
id=99999,
name="test-integration-storage",
password="test-integration-pass",
region="DE"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=88888,
name="test-integration-cdn",
hostname="test-integration.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Integration Site',
'--domain', 'test-integration.example.com',
'--storage-name', 'test-integration-storage',
'--region', 'DE',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert result.exit_code == 0
assert "Site provisioned successfully!" in result.output
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
site = deployment_repo.get_by_hostname('test-integration.example.com')
assert site is not None
assert site.site_name == 'Test Integration Site'
assert site.storage_zone_id == 99999
assert site.storage_zone_name == 'test-integration-storage'
assert site.storage_zone_password == 'test-integration-pass'
assert site.storage_zone_region == 'DE'
assert site.pull_zone_id == 88888
assert site.pull_zone_bcdn_hostname == 'test-integration.b-cdn.net'
finally:
session.close()
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
def test_provision_then_list_sites(
self, mock_client_class, mock_get_key, cli_runner, setup_test_admin, cleanup_test_sites
):
"""Test provisioning a site and then listing it"""
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.create_storage_zone.return_value = StorageZoneResult(
id=99999, name="test-list-storage", password="test-pass", region="NY"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=88888, name="test-list-cdn", hostname="test-list.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
provision_result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test List Site',
'--domain', 'test-list.example.com',
'--storage-name', 'test-list-storage',
'--region', 'NY',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert provision_result.exit_code == 0
list_result = cli_runner.invoke(app, [
'list-sites',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert list_result.exit_code == 0
assert "test-list.example.com" in list_result.output
assert "Test List Site" in list_result.output
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
def test_provision_get_and_remove_workflow(
self, mock_client_class, mock_get_key, cli_runner, setup_test_admin, cleanup_test_sites
):
"""Test complete workflow: provision, get details, then remove"""
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.create_storage_zone.return_value = StorageZoneResult(
id=77777, name="test-remove-storage", password="test-pass", region="LA"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=66666, name="test-remove-cdn", hostname="test-remove.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
provision_result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Remove Site',
'--domain', 'test-remove.example.com',
'--storage-name', 'test-remove-storage',
'--region', 'LA',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert provision_result.exit_code == 0
get_result = cli_runner.invoke(app, [
'get-site',
'--domain', 'test-remove.example.com',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert get_result.exit_code == 0
assert "Test Remove Site" in get_result.output
assert "77777" in get_result.output
assert "test-remove-storage" in get_result.output
remove_result = cli_runner.invoke(app, [
'remove-site',
'--domain', 'test-remove.example.com',
'--admin-user', 'test_admin',
'--admin-password', 'test_password',
'--yes'
])
assert remove_result.exit_code == 0
assert "has been removed" in remove_result.output
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
site = deployment_repo.get_by_hostname('test-remove.example.com')
assert site is None
finally:
session.close()
class TestAttachDomainIntegration:
"""Integration tests for attach-domain workflow"""
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
def test_attach_domain_to_existing_storage(
self, mock_client_class, mock_get_key, cli_runner, setup_test_admin, cleanup_test_sites
):
"""Test attaching a domain to an existing storage zone"""
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.find_storage_zone_by_name.return_value = StorageZoneResult(
id=55555,
name="existing-storage",
password="existing-pass",
region="SG"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=44444,
name="attach-test-cdn",
hostname="attach-test.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
result = cli_runner.invoke(app, [
'attach-domain',
'--name', 'Test Attach Site',
'--domain', 'test-attach.example.com',
'--storage-name', 'existing-storage',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert result.exit_code == 0
assert "Domain attached successfully!" in result.output
session = db_manager.get_session()
try:
deployment_repo = SiteDeploymentRepository(session)
site = deployment_repo.get_by_hostname('test-attach.example.com')
assert site is not None
assert site.site_name == 'Test Attach Site'
assert site.storage_zone_id == 55555
assert site.storage_zone_name == 'existing-storage'
assert site.pull_zone_id == 44444
finally:
session.close()
class TestDeploymentErrorHandling:
"""Integration tests for error handling in deployment workflow"""
def test_provision_without_api_key(self, cli_runner, setup_test_admin):
"""Test provisioning fails gracefully without API key"""
with patch('src.cli.commands.get_bunny_account_api_key', side_effect=ValueError("API key required")):
result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Site',
'--domain', 'test.example.com',
'--storage-name', 'test-storage',
'--region', 'DE',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert result.exit_code != 0
assert "BUNNY_ACCOUNT_API_KEY" in result.output
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
def test_provision_duplicate_domain(
self, mock_client_class, mock_get_key, cli_runner, setup_test_admin, cleanup_test_sites
):
"""Test provisioning fails when domain already exists"""
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.create_storage_zone.return_value = StorageZoneResult(
id=12345, name="test-storage", password="test-pass", region="DE"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=67890, name="test-cdn", hostname="test.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
first_result = cli_runner.invoke(app, [
'provision-site',
'--name', 'First Site',
'--domain', 'duplicate.example.com',
'--storage-name', 'test-dup-storage-1',
'--region', 'DE',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert first_result.exit_code == 0
second_result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Second Site',
'--domain', 'duplicate.example.com',
'--storage-name', 'test-dup-storage-2',
'--region', 'DE',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert second_result.exit_code != 0
assert "already exists" in second_result.output
def test_get_nonexistent_site(self, cli_runner, setup_test_admin):
"""Test getting details of non-existent site"""
result = cli_runner.invoke(app, [
'get-site',
'--domain', 'nonexistent.example.com',
'--admin-user', 'test_admin',
'--admin-password', 'test_password'
])
assert result.exit_code != 0
assert "not found" in result.output
def test_remove_nonexistent_site(self, cli_runner, setup_test_admin):
"""Test removing non-existent site"""
result = cli_runner.invoke(app, [
'remove-site',
'--domain', 'nonexistent.example.com',
'--admin-user', 'test_admin',
'--admin-password', 'test_password',
'--yes'
])
assert result.exit_code != 0
assert "not found" in result.output

View File

@ -0,0 +1,203 @@
"""
Unit tests for bunny.net API client
"""
import pytest
from unittest.mock import Mock, patch
import requests
from src.deployment.bunnynet import (
BunnyNetClient,
BunnyNetAPIError,
BunnyNetAuthError,
BunnyNetResourceConflictError,
StorageZoneResult,
PullZoneResult
)
@pytest.fixture
def bunny_client():
"""Fixture to create a BunnyNetClient instance"""
return BunnyNetClient(api_key="test_api_key")
@pytest.fixture
def mock_response():
"""Fixture to create a mock response"""
mock = Mock(spec=requests.Response)
mock.status_code = 200
mock.json.return_value = {}
mock.content = b'{}'
return mock
class TestBunnyNetClient:
"""Tests for BunnyNetClient"""
def test_client_initialization(self):
"""Test client initializes with correct headers"""
client = BunnyNetClient("test_key")
assert client.api_key == "test_key"
assert client.session.headers["AccessKey"] == "test_key"
assert client.session.headers["Content-Type"] == "application/json"
def test_create_storage_zone_success(self, bunny_client, mock_response):
"""Test successful storage zone creation"""
mock_response.json.return_value = {
"Id": 12345,
"Name": "test-storage",
"Password": "test-password",
"Region": "DE"
}
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.create_storage_zone("test-storage", "DE")
assert isinstance(result, StorageZoneResult)
assert result.id == 12345
assert result.name == "test-storage"
assert result.password == "test-password"
assert result.region == "DE"
def test_create_storage_zone_auth_error(self, bunny_client, mock_response):
"""Test storage zone creation with auth error"""
mock_response.status_code = 401
with patch.object(bunny_client.session, 'request', return_value=mock_response):
with pytest.raises(BunnyNetAuthError):
bunny_client.create_storage_zone("test-storage", "DE")
def test_create_storage_zone_conflict(self, bunny_client, mock_response):
"""Test storage zone creation with name conflict"""
mock_response.status_code = 409
with patch.object(bunny_client.session, 'request', return_value=mock_response):
with pytest.raises(BunnyNetResourceConflictError):
bunny_client.create_storage_zone("existing-storage", "DE")
def test_get_storage_zones_success(self, bunny_client, mock_response):
"""Test successful retrieval of storage zones"""
mock_response.json.return_value = [
{"Id": 1, "Name": "zone1", "Region": "DE"},
{"Id": 2, "Name": "zone2", "Region": "NY"}
]
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.get_storage_zones()
assert len(result) == 2
assert result[0]["Name"] == "zone1"
assert result[1]["Name"] == "zone2"
def test_find_storage_zone_by_name_found(self, bunny_client, mock_response):
"""Test finding an existing storage zone by name"""
mock_response.json.return_value = [
{"Id": 1, "Name": "zone1", "Region": "DE", "Password": "pass1"},
{"Id": 2, "Name": "zone2", "Region": "NY", "Password": "pass2"}
]
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.find_storage_zone_by_name("zone2")
assert result is not None
assert result.id == 2
assert result.name == "zone2"
assert result.region == "NY"
assert result.password == "pass2"
def test_find_storage_zone_by_name_not_found(self, bunny_client, mock_response):
"""Test finding a non-existent storage zone"""
mock_response.json.return_value = [
{"Id": 1, "Name": "zone1", "Region": "DE", "Password": "pass1"}
]
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.find_storage_zone_by_name("nonexistent")
assert result is None
def test_create_pull_zone_success(self, bunny_client, mock_response):
"""Test successful pull zone creation"""
mock_response.json.return_value = {
"Id": 67890,
"Name": "test-cdn",
"Hostnames": [{"Value": "test-cdn.b-cdn.net"}]
}
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.create_pull_zone("test-cdn", 12345)
assert isinstance(result, PullZoneResult)
assert result.id == 67890
assert result.name == "test-cdn"
assert result.hostname == "test-cdn.b-cdn.net"
def test_create_pull_zone_no_hostnames(self, bunny_client, mock_response):
"""Test pull zone creation with no hostnames in response"""
mock_response.json.return_value = {
"Id": 67890,
"Name": "test-cdn"
}
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.create_pull_zone("test-cdn", 12345)
assert result.hostname == "test-cdn.b-cdn.net"
def test_create_pull_zone_conflict(self, bunny_client, mock_response):
"""Test pull zone creation with name conflict"""
mock_response.status_code = 409
with patch.object(bunny_client.session, 'request', return_value=mock_response):
with pytest.raises(BunnyNetResourceConflictError):
bunny_client.create_pull_zone("existing-cdn", 12345)
def test_add_custom_hostname_success(self, bunny_client, mock_response):
"""Test successful custom hostname addition"""
mock_response.status_code = 200
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client.add_custom_hostname(67890, "www.example.com")
assert result is True
def test_add_custom_hostname_error(self, bunny_client, mock_response):
"""Test custom hostname addition with error"""
mock_response.status_code = 400
mock_response.json.return_value = {"Message": "Invalid hostname"}
with patch.object(bunny_client.session, 'request', return_value=mock_response):
with pytest.raises(BunnyNetAPIError):
bunny_client.add_custom_hostname(67890, "invalid")
def test_request_timeout(self, bunny_client):
"""Test handling of request timeout"""
with patch.object(bunny_client.session, 'request', side_effect=requests.exceptions.Timeout):
with pytest.raises(BunnyNetAPIError, match="timeout"):
bunny_client.create_storage_zone("test", "DE")
def test_request_connection_error(self, bunny_client):
"""Test handling of connection error"""
with patch.object(bunny_client.session, 'request', side_effect=requests.exceptions.ConnectionError):
with pytest.raises(BunnyNetAPIError, match="Connection error"):
bunny_client.create_storage_zone("test", "DE")
def test_api_error_with_message(self, bunny_client, mock_response):
"""Test API error handling with message in response"""
mock_response.status_code = 500
mock_response.json.return_value = {"Message": "Internal server error"}
with patch.object(bunny_client.session, 'request', return_value=mock_response):
with pytest.raises(BunnyNetAPIError, match="Internal server error"):
bunny_client.create_storage_zone("test", "DE")
def test_empty_response_handling(self, bunny_client, mock_response):
"""Test handling of empty response"""
mock_response.status_code = 204
mock_response.content = b''
with patch.object(bunny_client.session, 'request', return_value=mock_response):
result = bunny_client._make_request("POST", "/test")
assert result == {}

View File

@ -0,0 +1,411 @@
"""
Unit tests for deployment CLI commands
"""
import pytest
from click.testing import CliRunner
from unittest.mock import Mock, patch, MagicMock
from src.cli.commands import app
from src.database.models import User, SiteDeployment
from src.deployment.bunnynet import (
StorageZoneResult,
PullZoneResult,
BunnyNetAuthError,
BunnyNetResourceConflictError,
BunnyNetAPIError
)
@pytest.fixture
def cli_runner():
"""Fixture to create a Click CLI test runner"""
return CliRunner()
@pytest.fixture
def mock_admin_user():
"""Fixture for a mock admin user"""
user = Mock(spec=User)
user.id = 1
user.username = "admin"
user.role = "Admin"
user.is_admin.return_value = True
return user
@pytest.fixture
def mock_site_deployment():
"""Fixture for a mock site deployment"""
site = Mock(spec=SiteDeployment)
site.id = 1
site.site_name = "Test Site"
site.custom_hostname = "www.example.com"
site.storage_zone_id = 12345
site.storage_zone_name = "test-storage"
site.storage_zone_password = "test-pass"
site.storage_zone_region = "DE"
site.pull_zone_id = 67890
site.pull_zone_bcdn_hostname = "test-cdn.b-cdn.net"
site.created_at = Mock()
site.created_at.strftime.return_value = "2024-01-01 00:00:00"
site.updated_at = Mock()
site.updated_at.strftime.return_value = "2024-01-01 00:00:00"
return site
class TestProvisionSiteCommand:
"""Tests for provision-site CLI command"""
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
@patch('src.cli.commands.db_manager')
def test_provision_site_success(
self, mock_db, mock_client_class, mock_get_key, mock_auth, cli_runner, mock_admin_user
):
"""Test successful site provisioning"""
mock_auth.return_value = mock_admin_user
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.create_storage_zone.return_value = StorageZoneResult(
id=12345, name="test-storage", password="test-pass", region="DE"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=67890, name="test-cdn", hostname="test-cdn.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.exists.return_value = False
mock_repo.create.return_value = Mock()
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Site',
'--domain', 'www.example.com',
'--storage-name', 'test-storage',
'--region', 'DE',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code == 0
assert "Site provisioned successfully!" in result.output
assert "MANUAL DNS CONFIGURATION REQUIRED" in result.output
mock_client.create_storage_zone.assert_called_once_with("test-storage", "DE")
mock_client.create_pull_zone.assert_called_once()
mock_client.add_custom_hostname.assert_called_once()
@patch('src.cli.commands.authenticate_admin')
def test_provision_site_auth_failure(self, mock_auth, cli_runner):
"""Test provision-site with authentication failure"""
mock_auth.return_value = None
result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Site',
'--domain', 'www.example.com',
'--storage-name', 'test-storage',
'--region', 'DE',
'--admin-user', 'admin',
'--admin-password', 'wrong'
])
assert result.exit_code != 0
assert "Authentication failed" in result.output
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.get_bunny_account_api_key')
def test_provision_site_missing_api_key(self, mock_get_key, mock_auth, cli_runner, mock_admin_user):
"""Test provision-site with missing API key"""
mock_auth.return_value = mock_admin_user
mock_get_key.side_effect = ValueError("BUNNY_ACCOUNT_API_KEY environment variable is required")
result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Site',
'--domain', 'www.example.com',
'--storage-name', 'test-storage',
'--region', 'DE',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code != 0
assert "BUNNY_ACCOUNT_API_KEY" in result.output
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
@patch('src.cli.commands.db_manager')
def test_provision_site_domain_exists(
self, mock_db, mock_client_class, mock_get_key, mock_auth, cli_runner, mock_admin_user
):
"""Test provision-site with existing domain"""
mock_auth.return_value = mock_admin_user
mock_get_key.return_value = "test_api_key"
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.exists.return_value = True
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'provision-site',
'--name', 'Test Site',
'--domain', 'www.example.com',
'--storage-name', 'test-storage',
'--region', 'DE',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code != 0
assert "already exists" in result.output
class TestAttachDomainCommand:
"""Tests for attach-domain CLI command"""
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
@patch('src.cli.commands.db_manager')
def test_attach_domain_success(
self, mock_db, mock_client_class, mock_get_key, mock_auth, cli_runner, mock_admin_user
):
"""Test successful domain attachment"""
mock_auth.return_value = mock_admin_user
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.find_storage_zone_by_name.return_value = StorageZoneResult(
id=12345, name="existing-storage", password="test-pass", region="DE"
)
mock_client.create_pull_zone.return_value = PullZoneResult(
id=67890, name="test-cdn", hostname="test-cdn.b-cdn.net"
)
mock_client.add_custom_hostname.return_value = True
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.exists.return_value = False
mock_repo.create.return_value = Mock()
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'attach-domain',
'--name', 'Test Site',
'--domain', 'www.example.com',
'--storage-name', 'existing-storage',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code == 0
assert "Domain attached successfully!" in result.output
assert "MANUAL DNS CONFIGURATION REQUIRED" in result.output
mock_client.find_storage_zone_by_name.assert_called_once_with("existing-storage")
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.get_bunny_account_api_key')
@patch('src.cli.commands.BunnyNetClient')
@patch('src.cli.commands.db_manager')
def test_attach_domain_storage_not_found(
self, mock_db, mock_client_class, mock_get_key, mock_auth, cli_runner, mock_admin_user
):
"""Test attach-domain with non-existent storage zone"""
mock_auth.return_value = mock_admin_user
mock_get_key.return_value = "test_api_key"
mock_client = Mock()
mock_client_class.return_value = mock_client
mock_client.find_storage_zone_by_name.return_value = None
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.exists.return_value = False
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'attach-domain',
'--name', 'Test Site',
'--domain', 'www.example.com',
'--storage-name', 'nonexistent',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code != 0
assert "not found" in result.output
class TestListSitesCommand:
"""Tests for list-sites CLI command"""
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.db_manager')
def test_list_sites_success(self, mock_db, mock_auth, cli_runner, mock_admin_user, mock_site_deployment):
"""Test successful sites listing"""
mock_auth.return_value = mock_admin_user
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.get_all.return_value = [mock_site_deployment]
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'list-sites',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code == 0
assert "Total sites: 1" in result.output
assert "www.example.com" in result.output
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.db_manager')
def test_list_sites_empty(self, mock_db, mock_auth, cli_runner, mock_admin_user):
"""Test listing sites when none exist"""
mock_auth.return_value = mock_admin_user
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.get_all.return_value = []
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'list-sites',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code == 0
assert "No site deployments found" in result.output
class TestGetSiteCommand:
"""Tests for get-site CLI command"""
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.db_manager')
def test_get_site_success(self, mock_db, mock_auth, cli_runner, mock_admin_user, mock_site_deployment):
"""Test successful site details retrieval"""
mock_auth.return_value = mock_admin_user
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.get_by_hostname.return_value = mock_site_deployment
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'get-site',
'--domain', 'www.example.com',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code == 0
assert "Site Deployment Details" in result.output
assert "www.example.com" in result.output
assert "test-storage" in result.output
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.db_manager')
def test_get_site_not_found(self, mock_db, mock_auth, cli_runner, mock_admin_user):
"""Test get-site with non-existent domain"""
mock_auth.return_value = mock_admin_user
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.get_by_hostname.return_value = None
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'get-site',
'--domain', 'nonexistent.com',
'--admin-user', 'admin',
'--admin-password', 'password'
])
assert result.exit_code != 0
assert "not found" in result.output
class TestRemoveSiteCommand:
"""Tests for remove-site CLI command"""
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.db_manager')
def test_remove_site_success(self, mock_db, mock_auth, cli_runner, mock_admin_user, mock_site_deployment):
"""Test successful site removal"""
mock_auth.return_value = mock_admin_user
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.get_by_hostname.return_value = mock_site_deployment
mock_repo.delete.return_value = True
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'remove-site',
'--domain', 'www.example.com',
'--admin-user', 'admin',
'--admin-password', 'password',
'--yes'
])
assert result.exit_code == 0
assert "has been removed" in result.output
assert "does NOT delete resources from bunny.net" in result.output
@patch('src.cli.commands.authenticate_admin')
@patch('src.cli.commands.db_manager')
def test_remove_site_not_found(self, mock_db, mock_auth, cli_runner, mock_admin_user):
"""Test remove-site with non-existent domain"""
mock_auth.return_value = mock_admin_user
mock_session = Mock()
mock_db.get_session.return_value = mock_session
mock_repo = Mock()
mock_repo.get_by_hostname.return_value = None
with patch('src.cli.commands.SiteDeploymentRepository', return_value=mock_repo):
result = cli_runner.invoke(app, [
'remove-site',
'--domain', 'nonexistent.com',
'--admin-user', 'admin',
'--admin-password', 'password',
'--yes'
])
assert result.exit_code != 0
assert "not found" in result.output