sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: modernize code quality tooling with Ruff

Browse Source 
- Replace black, isort, and flake8 with Ruff (all-in-one linter and formatter)
- Add comprehensive pyproject.toml configuration
- Simplify Makefile code quality targets
- Configure exclusions for venv/.venv in pyproject.toml
- Auto-fix 1,359 linting issues across codebase

Benefits:
- Much faster builds (Ruff is written in Rust)
- Single tool replaces multiple tools
- More comprehensive rule set (UP, B, C4, SIM, PIE, RET, Q)
- All configuration centralized in pyproject.toml
- Better import sorting and formatting consistency

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-11-28 19:37:38 +01:00
parent 21c13ca39b
commit 238c1ec9b8
169 changed files with 2183 additions and 1784 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
scripts/backup_database.py
View File

@@ -2,12 +2,10 @@
"""Database backup utility that uses project configuration."""
import os
import shutil
import sqlite3
import sys
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse
# Add project root to Python path
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
@@ -25,11 +23,10 @@ def get_database_path():
if db_path.startswith("./"):
db_path = db_path[2:] # Remove ./ prefix
return db_path
else:
# For PostgreSQL or other databases, we can't do file backup
print(f"[INFO] Database type: {db_url.split('://')[0]}")
print("[ERROR] File backup only supported for SQLite databases")
return None
# For PostgreSQL or other databases, we can't do file backup
print(f"[INFO] Database type: {db_url.split('://')[0]}")
print("[ERROR] File backup only supported for SQLite databases")
return None
def backup_sqlite_database():
@@ -78,10 +75,9 @@ def backup_database():
if settings.database_url.startswith("sqlite"):
return backup_sqlite_database()
else:
print("[INFO] For PostgreSQL databases, use pg_dump:")
print(f"pg_dump {settings.database_url} > backup_$(date +%Y%m%d_%H%M%S).sql")
return True
print("[INFO] For PostgreSQL databases, use pg_dump:")
print(f"pg_dump {settings.database_url} > backup_$(date +%Y%m%d_%H%M%S).sql")
return True
if __name__ == "__main__":

10
scripts/create_default_content_pages.py
View File

@@ -26,7 +26,7 @@ Usage:
"""
import sys
from datetime import datetime, timezone
from datetime import UTC, datetime
from pathlib import Path
# Add project root to path
@@ -496,12 +496,12 @@ def create_default_pages(db: Session) -> None:
meta_description=page_data["meta_description"],
meta_keywords=page_data["meta_keywords"],
is_published=True,
published_at=datetime.now(timezone.utc),
published_at=datetime.now(UTC),
show_in_footer=page_data["show_in_footer"],
show_in_header=page_data.get("show_in_header", False),
display_order=page_data["display_order"],
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(page)
@@ -511,7 +511,7 @@ def create_default_pages(db: Session) -> None:
db.commit()
print("\n" + "=" * 70)
print(f"Summary:")
print("Summary:")
print(f" Created: {created_count} pages")
print(f" Skipped: {skipped_count} pages (already exist)")
print(f" Total: {created_count + skipped_count} pages")

14
scripts/create_landing_page.py
View File

@@ -12,7 +12,7 @@ from pathlib import Path
# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from datetime import datetime, timezone
from datetime import UTC, datetime
from sqlalchemy.orm import Session
@@ -65,7 +65,7 @@ def create_landing_page(
if content:
existing.content = content
existing.is_published = True
existing.updated_at = datetime.now(timezone.utc)
existing.updated_at = datetime.now(UTC)
db.commit()
print(f"✅ Updated landing page with template: {template}")
@@ -78,7 +78,7 @@ def create_landing_page(
content=content
or f"""
<h2>About {vendor.name}</h2>
<p>{vendor.description or 'Your trusted shopping destination for quality products.'}</p>
<p>{vendor.description or "Your trusted shopping destination for quality products."}</p>
<h3>Why Choose Us?</h3>
<ul>
@@ -95,7 +95,7 @@ def create_landing_page(
template=template,
meta_description=f"Shop at {vendor.name} for quality products and great service",
is_published=True,
published_at=datetime.now(timezone.utc),
published_at=datetime.now(UTC),
show_in_footer=False,
show_in_header=False,
display_order=0,
@@ -154,7 +154,7 @@ def list_vendors():
if landing:
print(f" Landing Page: ✅ ({landing.template})")
else:
print(f" Landing Page: ❌ None")
print(" Landing Page: ❌ None")
print()
finally:
@@ -220,8 +220,8 @@ if __name__ == "__main__":
if success:
print("\n✅ SUCCESS! Landing page is ready.")
print("\n💡 Try different templates:")
print(f" python scripts/create_landing_page.py")
print(f" # Then choose: minimal, modern, or full")
print(" python scripts/create_landing_page.py")
print(" # Then choose: minimal, modern, or full")
else:
print("\n❌ Failed to create landing page")
sys.exit(1)

23
scripts/init_production.py
View File

@@ -17,7 +17,7 @@ This script is idempotent - safe to run multiple times.
"""
import sys
from datetime import datetime, timezone
from datetime import UTC, datetime
from pathlib import Path
# Add project root to path
@@ -27,10 +27,13 @@ sys.path.insert(0, str(project_root))
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.config import (print_environment_info, settings,
validate_production_settings)
from app.core.config import (
print_environment_info,
settings,
validate_production_settings,
)
from app.core.database import SessionLocal
from app.core.environment import get_environment, is_production
from app.core.environment import is_production
from app.core.permissions import PermissionGroups
from middleware.auth import AuthManager
from models.database.admin import AdminSetting
@@ -97,8 +100,8 @@ def create_admin_user(db: Session, auth_manager: AuthManager) -> User:
last_name=settings.admin_last_name,
is_active=True,
is_email_verified=True,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(admin)
@@ -218,8 +221,8 @@ def create_admin_settings(db: Session) -> int:
value_type=setting_data["value_type"],
description=setting_data.get("description"),
is_public=setting_data.get("is_public", False),
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(setting)
settings_created += 1
@@ -320,14 +323,14 @@ def print_summary(db: Session):
user_count = db.query(User).filter(User.role == "admin").count()
setting_count = db.query(AdminSetting).count()
print(f"\n📊 Database Status:")
print("\n📊 Database Status:")
print(f" Admin users: {user_count}")
print(f" Admin settings: {setting_count}")
print("\n" + "─" * 70)
print("🔐 ADMIN CREDENTIALS")
print("─" * 70)
print(f" URL: /admin/login")
print(" URL: /admin/login")
print(f" Username: {settings.admin_username}")
print(f" Password: {settings.admin_password}")
print("─" * 70)

11
scripts/route_diagnostics.py
View File

@@ -6,8 +6,6 @@ Run this script to check if your vendor routes are properly configured.
Usage: python route_diagnostics.py
"""
import sys
from typing import Dict, List
def check_route_order():
@@ -58,7 +56,7 @@ def check_route_order():
vendor_info_found = False
for route in routes:
if hasattr(route, "path"):
if "/{vendor_code}" == route.path and "GET" in getattr(
if route.path == "/{vendor_code}" and "GET" in getattr(
route, "methods", set()
):
vendor_info_found = True
@@ -109,14 +107,13 @@ def test_vendor_endpoint():
print(f" Vendor: {data.get('name', 'N/A')}")
print(f" Code: {data.get('vendor_code', 'N/A')}")
return True
elif "text/html" in content_type:
if "text/html" in content_type:
print("❌ ERROR: Response is HTML, not JSON!")
print(" This confirms the route ordering issue")
print(" The HTML page route is catching the API request")
return False
else:
print(f"⚠️ Unknown content type: {content_type}")
return False
print(f"⚠️ Unknown content type: {content_type}")
return False
except requests.exceptions.ConnectionError:
print("⚠️ Cannot connect to server. Is FastAPI running on localhost:8000?")

63
scripts/seed_demo.py
View File

@@ -26,10 +26,9 @@ This script is idempotent when run normally.
"""
import sys
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from typing import Dict, List
# Add project root to path
project_root = Path(__file__).parent.parent
@@ -260,7 +259,7 @@ def reset_all_data(db: Session):
# =============================================================================
def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
def create_demo_vendors(db: Session, auth_manager: AuthManager) -> list[Vendor]:
"""Create demo vendors with users."""
vendors = []
@@ -291,8 +290,8 @@ def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
last_name=user_data["last_name"],
is_active=True,
is_email_verified=True,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(vendor_user)
db.flush()
@@ -305,8 +304,8 @@ def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
description=vendor_data["description"],
is_active=True,
is_verified=True,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(vendor)
db.flush()
@@ -317,7 +316,7 @@ def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
user_id=vendor_user.id,
user_type="owner",
is_active=True,
created_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
)
db.add(vendor_user_link)
@@ -334,8 +333,8 @@ def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
background_color=theme_colors["background"],
text_color=theme_colors["text"],
is_active=True,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(theme)
@@ -347,8 +346,8 @@ def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
is_verified=True, # Auto-verified for demo
is_active=True,
verification_token=None,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(domain)
@@ -361,7 +360,7 @@ def create_demo_vendors(db: Session, auth_manager: AuthManager) -> List[Vendor]:
def create_demo_customers(
db: Session, vendor: Vendor, auth_manager: AuthManager, count: int
) -> List[Customer]:
) -> list[Customer]:
"""Create demo customers for a vendor."""
customers = []
@@ -388,12 +387,12 @@ def create_demo_customers(
email=email,
hashed_password=auth_manager.hash_password(demo_password),
first_name=f"Customer{i}",
last_name=f"Test",
last_name="Test",
phone=f"+352123456{i:03d}",
customer_number=customer_number,
is_active=True,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(customer)
customers.append(customer)
@@ -409,7 +408,7 @@ def create_demo_customers(
return customers
def create_demo_products(db: Session, vendor: Vendor, count: int) -> List[Product]:
def create_demo_products(db: Session, vendor: Vendor, count: int) -> list[Product]:
"""Create demo products for a vendor."""
products = []
@@ -455,8 +454,8 @@ def create_demo_products(db: Session, vendor: Vendor, count: int) -> List[Produc
marketplace="Wizamart",
vendor_name=vendor.name,
currency="EUR",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(marketplace_product)
db.flush() # Flush to get the marketplace_product.id
@@ -474,8 +473,8 @@ def create_demo_products(db: Session, vendor: Vendor, count: int) -> List[Produc
is_featured=(i % 5 == 0), # Every 5th product is featured
display_order=i,
min_quantity=1,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(product)
products.append(product)
@@ -548,7 +547,7 @@ def print_summary(db: Session):
customer_count = db.query(Customer).count()
product_count = db.query(Product).count()
print(f"\n📊 Database Status:")
print("\n📊 Database Status:")
print(f" Vendors: {vendor_count}")
print(f" Users: {user_count}")
print(f" Customers: {customer_count}")
@@ -556,7 +555,7 @@ def print_summary(db: Session):
# Show vendor details
vendors = db.query(Vendor).all()
print(f"\n🏪 Demo Vendors:")
print("\n🏪 Demo Vendors:")
for vendor in vendors:
print(f"\n {vendor.name} ({vendor.vendor_code})")
print(f" Subdomain: {vendor.subdomain}.{settings.platform_domain}")
@@ -580,7 +579,7 @@ def print_summary(db: Session):
print(f" Status: {'✓ Active' if vendor.is_active else '✗ Inactive'}")
print(f"\n🔐 Demo Vendor Credentials:")
print("\n🔐 Demo Vendor Credentials:")
print("─" * 70)
for i, vendor_data in enumerate(DEMO_VENDOR_USERS[:vendor_count], 1):
vendor = vendors[i - 1] if i <= len(vendors) else None
@@ -597,15 +596,15 @@ def print_summary(db: Session):
)
print()
print(f"\n🛒 Demo Customer Credentials:")
print("\n🛒 Demo Customer Credentials:")
print("─" * 70)
print(f" All customers:")
print(f" Email: customer1@{{subdomain}}.example.com")
print(f" Password: customer123")
print(f" (Replace {{subdomain}} with vendor subdomain, e.g., wizamart)")
print(" All customers:")
print(" Email: customer1@{subdomain}.example.com")
print(" Password: customer123")
print(" (Replace {subdomain} with vendor subdomain, e.g., wizamart)")
print()
print(f"\n🏪 Shop Access (Development):")
print("\n🏪 Shop Access (Development):")
print("─" * 70)
for vendor in vendors:
print(f" {vendor.name}:")
@@ -622,8 +621,8 @@ def print_summary(db: Session):
print(" 2. Login as vendor:")
print(" • Path-based: http://localhost:8000/vendor/WIZAMART/login")
print(" • Subdomain: http://wizamart.localhost:8000/vendor/login")
print(f" 3. Visit vendor shop: http://localhost:8000/vendors/WIZAMART/shop/")
print(f" 4. Admin panel: http://localhost:8000/admin/login")
print(" 3. Visit vendor shop: http://localhost:8000/vendors/WIZAMART/shop/")
print(" 4. Admin panel: http://localhost:8000/admin/login")
print(f" Username: {settings.admin_username}")
print(f" Password: {settings.admin_password}")

30
scripts/show_structure.py
View File

@@ -15,7 +15,6 @@ import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List
def count_files(directory: str, pattern: str) -> int:
@@ -38,7 +37,7 @@ def count_files(directory: str, pattern: str) -> int:
return count
def get_tree_structure(directory: str, exclude_patterns: List[str] = None) -> str:
def get_tree_structure(directory: str, exclude_patterns: list[str] = None) -> str:
"""Generate tree structure for directory."""
if not os.path.exists(directory):
return f"Directory {directory} not found"
@@ -55,19 +54,18 @@ def get_tree_structure(directory: str, exclude_patterns: List[str] = None) -> st
errors="replace",
)
return result.stdout
else:
# Linux/Mac tree command with exclusions
exclude_args = []
if exclude_patterns:
exclude_args = ["-I", "|".join(exclude_patterns)]
# Linux/Mac tree command with exclusions
exclude_args = []
if exclude_patterns:
exclude_args = ["-I", "|".join(exclude_patterns)]
result = subprocess.run(
["tree", "-F", "-a"] + exclude_args + [directory],
capture_output=True,
text=True,
)
if result.returncode == 0:
return result.stdout
result = subprocess.run(
["tree", "-F", "-a"] + exclude_args + [directory],
capture_output=True,
text=True,
)
if result.returncode == 0:
return result.stdout
except (subprocess.SubprocessError, FileNotFoundError):
pass
@@ -76,7 +74,7 @@ def get_tree_structure(directory: str, exclude_patterns: List[str] = None) -> st
def generate_manual_tree(
directory: str, exclude_patterns: List[str] = None, prefix: str = ""
directory: str, exclude_patterns: list[str] = None, prefix: str = ""
) -> str:
"""Generate tree structure manually when tree command is not available."""
if exclude_patterns is None:
@@ -567,7 +565,7 @@ def generate_test_structure() -> str:
if file.startswith("test_") and file.endswith(".py"):
filepath = os.path.join(root, file)
try:
with open(filepath, "r", encoding="utf-8") as f:
with open(filepath, encoding="utf-8") as f:
for line in f:
if line.strip().startswith("def test_"):
test_function_count += 1

83
scripts/test_auth_complete.py
View File

@@ -20,8 +20,6 @@ Requirements:
* Customer: username=customer, password=customer123, vendor_id=1
"""
import json
from typing import Dict, Optional
import requests
@@ -78,7 +76,7 @@ def print_warning(message: str):
# ============================================================================
def test_admin_login() -> Optional[Dict]:
def test_admin_login() -> dict | None:
"""Test admin login and cookie configuration"""
print_test("Admin Login")
@@ -106,10 +104,9 @@ def test_admin_login() -> Optional[Dict]:
print_error("admin_token cookie NOT set")
return {"token": data["access_token"], "user": data.get("user", {})}
else:
print_error(f"Login failed: {response.status_code}")
print_error(f"Response: {response.text}")
return None
print_error(f"Login failed: {response.status_code}")
print_error(f"Response: {response.text}")
return None
except Exception as e:
print_error(f"Exception during admin login: {str(e)}")
@@ -131,12 +128,11 @@ def test_admin_cannot_access_vendor_api(admin_token: str):
print_success("Admin correctly blocked from vendor API")
print_success(f"Error code: {data.get('error_code', 'N/A')}")
return True
elif response.status_code == 200:
if response.status_code == 200:
print_error("SECURITY ISSUE: Admin can access vendor API!")
return False
else:
print_warning(f"Unexpected status code: {response.status_code}")
return False
print_warning(f"Unexpected status code: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -157,12 +153,11 @@ def test_admin_cannot_access_customer_api(admin_token: str):
if response.status_code in [401, 403]:
print_success("Admin correctly blocked from customer pages")
return True
elif response.status_code == 200:
if response.status_code == 200:
print_error("SECURITY ISSUE: Admin can access customer pages!")
return False
else:
print_warning(f"Unexpected status code: {response.status_code}")
return False
print_warning(f"Unexpected status code: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -174,7 +169,7 @@ def test_admin_cannot_access_customer_api(admin_token: str):
# ============================================================================
def test_vendor_login() -> Optional[Dict]:
def test_vendor_login() -> dict | None:
"""Test vendor login and cookie configuration"""
print_test("Vendor Login")
@@ -209,10 +204,9 @@ def test_vendor_login() -> Optional[Dict]:
"user": data.get("user", {}),
"vendor": data.get("vendor", {}),
}
else:
print_error(f"Login failed: {response.status_code}")
print_error(f"Response: {response.text}")
return None
print_error(f"Login failed: {response.status_code}")
print_error(f"Response: {response.text}")
return None
except Exception as e:
print_error(f"Exception during vendor login: {str(e)}")
@@ -234,12 +228,11 @@ def test_vendor_cannot_access_admin_api(vendor_token: str):
print_success("Vendor correctly blocked from admin API")
print_success(f"Error code: {data.get('error_code', 'N/A')}")
return True
elif response.status_code == 200:
if response.status_code == 200:
print_error("SECURITY ISSUE: Vendor can access admin API!")
return False
else:
print_warning(f"Unexpected status code: {response.status_code}")
return False
print_warning(f"Unexpected status code: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -259,12 +252,11 @@ def test_vendor_cannot_access_customer_api(vendor_token: str):
if response.status_code in [401, 403]:
print_success("Vendor correctly blocked from customer pages")
return True
elif response.status_code == 200:
if response.status_code == 200:
print_error("SECURITY ISSUE: Vendor can access customer pages!")
return False
else:
print_warning(f"Unexpected status code: {response.status_code}")
return False
print_warning(f"Unexpected status code: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -276,7 +268,7 @@ def test_vendor_cannot_access_customer_api(vendor_token: str):
# ============================================================================
def test_customer_login() -> Optional[Dict]:
def test_customer_login() -> dict | None:
"""Test customer login and cookie configuration"""
print_test("Customer Login")
@@ -304,10 +296,9 @@ def test_customer_login() -> Optional[Dict]:
print_error("customer_token cookie NOT set")
return {"token": data["access_token"], "user": data.get("user", {})}
else:
print_error(f"Login failed: {response.status_code}")
print_error(f"Response: {response.text}")
return None
print_error(f"Login failed: {response.status_code}")
print_error(f"Response: {response.text}")
return None
except Exception as e:
print_error(f"Exception during customer login: {str(e)}")
@@ -329,12 +320,11 @@ def test_customer_cannot_access_admin_api(customer_token: str):
print_success("Customer correctly blocked from admin API")
print_success(f"Error code: {data.get('error_code', 'N/A')}")
return True
elif response.status_code == 200:
if response.status_code == 200:
print_error("SECURITY ISSUE: Customer can access admin API!")
return False
else:
print_warning(f"Unexpected status code: {response.status_code}")
return False
print_warning(f"Unexpected status code: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -356,12 +346,11 @@ def test_customer_cannot_access_vendor_api(customer_token: str):
print_success("Customer correctly blocked from vendor API")
print_success(f"Error code: {data.get('error_code', 'N/A')}")
return True
elif response.status_code == 200:
if response.status_code == 200:
print_error("SECURITY ISSUE: Customer can access vendor API!")
return False
else:
print_warning(f"Unexpected status code: {response.status_code}")
return False
print_warning(f"Unexpected status code: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -378,9 +367,8 @@ def test_public_shop_access():
if response.status_code == 200:
print_success("Public shop pages accessible without auth")
return True
else:
print_error(f"Failed to access public shop: {response.status_code}")
return False
print_error(f"Failed to access public shop: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -399,9 +387,8 @@ def test_health_check():
print_success("Health check passed")
print_info(f"Status: {data.get('status', 'N/A')}")
return True
else:
print_error(f"Health check failed: {response.status_code}")
return False
print_error(f"Health check failed: {response.status_code}")
return False
except Exception as e:
print_error(f"Exception: {str(e)}")
@@ -416,7 +403,7 @@ def test_health_check():
def main():
"""Run all tests"""
print(f"\n{Color.BOLD}{Color.CYAN}{'═' * 60}")
print(f" 🔒 COMPLETE AUTHENTICATION SYSTEM TEST SUITE")
print(" 🔒 COMPLETE AUTHENTICATION SYSTEM TEST SUITE")
print(f"{'═' * 60}{Color.END}")
print(f"Testing server at: {BASE_URL}")

64
scripts/test_vendor_management.py
View File

@@ -9,8 +9,6 @@ Tests:
- Transfer ownership
"""
import json
from pprint import pprint
import requests
@@ -35,10 +33,9 @@ def login_admin():
ADMIN_TOKEN = data["access_token"]
print("✅ Admin login successful")
return True
else:
print(f"❌ Admin login failed: {response.status_code}")
print(response.text)
return False
print(f"❌ Admin login failed: {response.status_code}")
print(response.text)
return False
def get_headers():
@@ -71,18 +68,17 @@ def test_create_vendor_with_both_emails():
if response.status_code == 200:
data = response.json()
print("✅ Vendor created successfully")
print(f"\n📧 Emails:")
print("\n📧 Emails:")
print(f" Owner Email: {data['owner_email']}")
print(f" Contact Email: {data['contact_email']}")
print(f"\n🔑 Credentials:")
print("\n🔑 Credentials:")
print(f" Username: {data['owner_username']}")
print(f" Password: {data['temporary_password']}")
print(f"\n🔗 Login URL: {data['login_url']}")
return data["id"]
else:
print(f"❌ Failed: {response.status_code}")
print(response.text)
return None
print(f"❌ Failed: {response.status_code}")
print(response.text)
return None
def test_create_vendor_single_email():
@@ -106,7 +102,7 @@ def test_create_vendor_single_email():
if response.status_code == 200:
data = response.json()
print("✅ Vendor created successfully")
print(f"\n📧 Emails:")
print("\n📧 Emails:")
print(f" Owner Email: {data['owner_email']}")
print(f" Contact Email: {data['contact_email']}")
@@ -116,10 +112,9 @@ def test_create_vendor_single_email():
print("❌ Contact email should have defaulted to owner email")
return data["id"]
else:
print(f"❌ Failed: {response.status_code}")
print(response.text)
return None
print(f"❌ Failed: {response.status_code}")
print(response.text)
return None
def test_update_vendor_contact_email(vendor_id):
@@ -142,15 +137,14 @@ def test_update_vendor_contact_email(vendor_id):
if response.status_code == 200:
data = response.json()
print("✅ Vendor updated successfully")
print(f"\n📧 Emails after update:")
print("\n📧 Emails after update:")
print(f" Owner Email: {data['owner_email']} (unchanged)")
print(f" Contact Email: {data['contact_email']} (updated)")
print(f"\n📝 Name: {data['name']} (updated)")
return True
else:
print(f"❌ Failed: {response.status_code}")
print(response.text)
return False
print(f"❌ Failed: {response.status_code}")
print(response.text)
return False
def test_get_vendor_details(vendor_id):
@@ -166,22 +160,21 @@ def test_get_vendor_details(vendor_id):
if response.status_code == 200:
data = response.json()
print("✅ Vendor details retrieved")
print(f"\n📋 Vendor Info:")
print("\n📋 Vendor Info:")
print(f" ID: {data['id']}")
print(f" Code: {data['vendor_code']}")
print(f" Name: {data['name']}")
print(f" Subdomain: {data['subdomain']}")
print(f"\n👤 Owner Info:")
print("\n👤 Owner Info:")
print(f" Username: {data['owner_username']}")
print(f" Email: {data['owner_email']}")
print(f"\n📧 Contact Info:")
print("\n📧 Contact Info:")
print(f" Email: {data['contact_email']}")
print(f" Phone: {data.get('contact_phone', 'N/A')}")
return True
else:
print(f"❌ Failed: {response.status_code}")
print(response.text)
return False
print(f"❌ Failed: {response.status_code}")
print(response.text)
return False
def test_transfer_ownership(vendor_id, new_owner_user_id):
@@ -197,7 +190,7 @@ def test_transfer_ownership(vendor_id, new_owner_user_id):
if response.status_code == 200:
current_data = response.json()
print(f"\n📋 Current Owner:")
print("\n📋 Current Owner:")
print(f" User ID: {current_data['owner_user_id']}")
print(f" Username: {current_data['owner_username']}")
print(f" Email: {current_data['owner_email']}")
@@ -218,21 +211,20 @@ def test_transfer_ownership(vendor_id, new_owner_user_id):
if response.status_code == 200:
data = response.json()
print(f"\n✅ {data['message']}")
print(f"\n👤 Old Owner:")
print("\n👤 Old Owner:")
print(f" ID: {data['old_owner']['id']}")
print(f" Username: {data['old_owner']['username']}")
print(f" Email: {data['old_owner']['email']}")
print(f"\n👤 New Owner:")
print("\n👤 New Owner:")
print(f" ID: {data['new_owner']['id']}")
print(f" Username: {data['new_owner']['username']}")
print(f" Email: {data['new_owner']['email']}")
print(f"\n📝 Reason: {data['transfer_reason']}")
print(f"⏰ Transferred at: {data['transferred_at']}")
return True
else:
print(f"❌ Failed: {response.status_code}")
print(response.text)
return False
print(f"❌ Failed: {response.status_code}")
print(response.text)
return False
def main():

33
scripts/validate_architecture.py
View File

@@ -25,7 +25,7 @@ import sys
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Tuple
from typing import Any
import yaml
@@ -56,7 +56,7 @@ class Violation:
class ValidationResult:
"""Results of architecture validation"""
violations: List[Violation] = field(default_factory=list)
violations: list[Violation] = field(default_factory=list)
files_checked: int = 0
rules_applied: int = 0
@@ -80,13 +80,13 @@ class ArchitectureValidator:
self.result = ValidationResult()
self.project_root = Path.cwd()
def _load_config(self) -> Dict[str, Any]:
def _load_config(self) -> dict[str, Any]:
"""Load validation rules from YAML config"""
if not self.config_path.exists():
print(f"❌ Configuration file not found: {self.config_path}")
sys.exit(1)
with open(self.config_path, "r") as f:
with open(self.config_path) as f:
config = yaml.safe_load(f)
print(f"📋 Loaded architecture rules: {config.get('project', 'unknown')}")
@@ -144,7 +144,7 @@ class ArchitectureValidator:
# API-004: Check authentication
self._check_endpoint_authentication(file_path, content, lines)
def _check_pydantic_usage(self, file_path: Path, content: str, lines: List[str]):
def _check_pydantic_usage(self, file_path: Path, content: str, lines: list[str]):
"""API-001: Ensure endpoints use Pydantic models"""
rule = self._get_rule("API-001")
if not rule:
@@ -180,7 +180,7 @@ class ArchitectureValidator:
)
def _check_no_business_logic_in_endpoints(
self, file_path: Path, content: str, lines: List[str]
self, file_path: Path, content: str, lines: list[str]
):
"""API-002: Ensure no business logic in endpoints"""
rule = self._get_rule("API-002")
@@ -213,7 +213,7 @@ class ArchitectureValidator:
)
def _check_endpoint_exception_handling(
self, file_path: Path, content: str, lines: List[str]
self, file_path: Path, content: str, lines: list[str]
):
"""API-003: Check proper exception handling in endpoints"""
rule = self._get_rule("API-003")
@@ -263,7 +263,7 @@ class ArchitectureValidator:
)
def _check_endpoint_authentication(
self, file_path: Path, content: str, lines: List[str]
self, file_path: Path, content: str, lines: list[str]
):
"""API-004: Check authentication on endpoints"""
rule = self._get_rule("API-004")
@@ -321,7 +321,7 @@ class ArchitectureValidator:
self._check_db_session_parameter(file_path, content, lines)
def _check_no_http_exception_in_services(
self, file_path: Path, content: str, lines: List[str]
self, file_path: Path, content: str, lines: list[str]
):
"""SVC-001: Services must not raise HTTPException"""
rule = self._get_rule("SVC-001")
@@ -357,7 +357,7 @@ class ArchitectureValidator:
)
def _check_service_exceptions(
self, file_path: Path, content: str, lines: List[str]
self, file_path: Path, content: str, lines: list[str]
):
"""SVC-002: Check for proper exception handling"""
rule = self._get_rule("SVC-002")
@@ -379,7 +379,7 @@ class ArchitectureValidator:
)
def _check_db_session_parameter(
self, file_path: Path, content: str, lines: List[str]
self, file_path: Path, content: str, lines: list[str]
):
"""SVC-003: Service methods should accept db session as parameter"""
rule = self._get_rule("SVC-003")
@@ -536,7 +536,7 @@ class ArchitectureValidator:
suggestion="Add {% extends 'admin/base.html' %} at the top",
)
def _get_rule(self, rule_id: str) -> Dict[str, Any]:
def _get_rule(self, rule_id: str) -> dict[str, Any]:
"""Get rule configuration by ID"""
# Look in different rule categories
for category in [
@@ -618,14 +618,13 @@ class ArchitectureValidator:
print("❌ VALIDATION FAILED - Fix errors before committing")
print("=" * 80)
return 1
elif self.result.has_warnings():
if self.result.has_warnings():
print("⚠️ VALIDATION PASSED WITH WARNINGS")
print("=" * 80)
return 0
else:
print("✅ VALIDATION PASSED - No violations found")
print("=" * 80)
return 0
print("✅ VALIDATION PASSED - No violations found")
print("=" * 80)
return 0
def print_json(self) -> int:
"""Print validation results as JSON"""

11
scripts/verify_setup.py
View File

@@ -5,10 +5,10 @@ import os
import sys
from urllib.parse import urlparse
from alembic.config import Config
from sqlalchemy import create_engine, text
from alembic import command
from alembic.config import Config
# Add project root to Python path
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
@@ -28,8 +28,7 @@ def get_database_info():
if db_path.startswith("./"):
db_path = db_path[2:]
return db_type, db_path
else:
return db_type, None
return db_type, None
def verify_database_setup():
@@ -138,7 +137,7 @@ def verify_model_structure():
try:
from models.database.base import Base
print(f"[OK] Database Base imported")
print("[OK] Database Base imported")
print(
f"[INFO] Found {len(Base.metadata.tables)} database tables: {list(Base.metadata.tables.keys())}"
)
@@ -191,7 +190,7 @@ def verify_model_structure():
def check_project_structure():
"""Check overall project structure."""
print(f"\n[STRUCTURE] Checking project structure...")
print("\n[STRUCTURE] Checking project structure...")
critical_paths = [
"models/database/base.py",
@@ -216,7 +215,7 @@ def check_project_structure():
"models/api/__init__.py",
]
print(f"\n[INIT] Checking __init__.py files...")
print("\n[INIT] Checking __init__.py files...")
for init_file in init_files:
if os.path.exists(init_file):
print(f" * {init_file}")