360 lines
14 KiB
Python
360 lines
14 KiB
Python
# tests/test_admin_service.py
|
|
import pytest
|
|
from datetime import datetime
|
|
from fastapi import HTTPException
|
|
|
|
from app.services.admin_service import AdminService
|
|
from models.database_models import User, Shop, MarketplaceImportJob
|
|
|
|
|
|
class TestAdminService:
|
|
"""Test suite for AdminService following the application's testing patterns"""
|
|
|
|
def setup_method(self):
|
|
"""Setup method following the same pattern as product service tests"""
|
|
self.service = AdminService()
|
|
|
|
def test_get_all_users(self, db, test_user, test_admin):
|
|
"""Test getting all users with pagination"""
|
|
users = self.service.get_all_users(db, skip=0, limit=10)
|
|
|
|
assert len(users) >= 2 # test_user + test_admin
|
|
user_ids = [user.id for user in users]
|
|
assert test_user.id in user_ids
|
|
assert test_admin.id in user_ids
|
|
|
|
def test_get_all_users_with_pagination(self, db, test_user, test_admin):
|
|
"""Test user pagination works correctly"""
|
|
users = self.service.get_all_users(db, skip=0, limit=1)
|
|
|
|
assert len(users) == 1
|
|
|
|
users_second_page = self.service.get_all_users(db, skip=1, limit=1)
|
|
assert len(users_second_page) == 1
|
|
assert users[0].id != users_second_page[0].id
|
|
|
|
def test_toggle_user_status_deactivate(self, db, test_user, test_admin):
|
|
"""Test deactivating a user"""
|
|
assert test_user.is_active is True
|
|
|
|
user, message = self.service.toggle_user_status(db, test_user.id, test_admin.id)
|
|
|
|
assert user.id == test_user.id
|
|
assert user.is_active is False
|
|
assert f"{user.username} has been deactivated" in message
|
|
|
|
def test_toggle_user_status_activate(self, db, test_user, test_admin):
|
|
"""Test activating a user"""
|
|
# First deactivate the user
|
|
test_user.is_active = False
|
|
db.commit()
|
|
|
|
user, message = self.service.toggle_user_status(db, test_user.id, test_admin.id)
|
|
|
|
assert user.id == test_user.id
|
|
assert user.is_active is True
|
|
assert f"{user.username} has been activated" in message
|
|
|
|
def test_toggle_user_status_user_not_found(self, db, test_admin):
|
|
"""Test toggle user status when user not found"""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
self.service.toggle_user_status(db, 99999, test_admin.id)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "User not found" in str(exc_info.value.detail)
|
|
|
|
def test_toggle_user_status_cannot_deactivate_self(self, db, test_admin):
|
|
"""Test that admin cannot deactivate their own account"""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
self.service.toggle_user_status(db, test_admin.id, test_admin.id)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "Cannot deactivate your own account" in str(exc_info.value.detail)
|
|
|
|
def test_get_all_shops(self, db, test_shop):
|
|
"""Test getting all shops with total count"""
|
|
shops, total = self.service.get_all_shops(db, skip=0, limit=10)
|
|
|
|
assert total >= 1
|
|
assert len(shops) >= 1
|
|
shop_codes = [shop.shop_code for shop in shops]
|
|
assert test_shop.shop_code in shop_codes
|
|
|
|
def test_get_all_shops_with_pagination(self, db, test_shop):
|
|
"""Test shop pagination works correctly"""
|
|
# Create additional shop for pagination test using the helper function
|
|
from conftest import create_test_import_job # If you added the helper function
|
|
|
|
# Or create directly with proper fields
|
|
additional_shop = Shop(
|
|
shop_code=f"{test_shop.shop_code}_2",
|
|
shop_name="Test Shop 2",
|
|
owner_id=test_shop.owner_id,
|
|
is_active=True,
|
|
is_verified=False
|
|
)
|
|
db.add(additional_shop)
|
|
db.commit()
|
|
|
|
shops_page_1 = self.service.get_all_shops(db, skip=0, limit=1)
|
|
assert len(shops_page_1[0]) == 1
|
|
|
|
shops_page_2 = self.service.get_all_shops(db, skip=1, limit=1)
|
|
assert len(shops_page_2[0]) == 1
|
|
|
|
# Ensure different shops on different pages
|
|
assert shops_page_1[0][0].id != shops_page_2[0][0].id
|
|
|
|
def test_verify_shop_mark_verified(self, db, test_shop):
|
|
"""Test marking shop as verified"""
|
|
# Ensure shop starts unverified
|
|
test_shop.is_verified = False
|
|
db.commit()
|
|
|
|
shop, message = self.service.verify_shop(db, test_shop.id)
|
|
|
|
assert shop.id == test_shop.id
|
|
assert shop.is_verified is True
|
|
assert f"{shop.shop_code} has been verified" in message
|
|
|
|
def test_verify_shop_mark_unverified(self, db, test_shop):
|
|
"""Test marking shop as unverified"""
|
|
# Ensure shop starts verified
|
|
test_shop.is_verified = True
|
|
db.commit()
|
|
|
|
shop, message = self.service.verify_shop(db, test_shop.id)
|
|
|
|
assert shop.id == test_shop.id
|
|
assert shop.is_verified is False
|
|
assert f"{shop.shop_code} has been unverified" in message
|
|
|
|
def test_verify_shop_not_found(self, db):
|
|
"""Test verify shop when shop not found"""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
self.service.verify_shop(db, 99999)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "Shop not found" in str(exc_info.value.detail)
|
|
|
|
def test_toggle_shop_status_deactivate(self, db, test_shop):
|
|
"""Test deactivating a shop"""
|
|
assert test_shop.is_active is True
|
|
|
|
shop, message = self.service.toggle_shop_status(db, test_shop.id)
|
|
|
|
assert shop.id == test_shop.id
|
|
assert shop.is_active is False
|
|
assert f"{shop.shop_code} has been deactivated" in message
|
|
|
|
def test_toggle_shop_status_activate(self, db, test_shop):
|
|
"""Test activating a shop"""
|
|
# First deactivate the shop
|
|
test_shop.is_active = False
|
|
db.commit()
|
|
|
|
shop, message = self.service.toggle_shop_status(db, test_shop.id)
|
|
|
|
assert shop.id == test_shop.id
|
|
assert shop.is_active is True
|
|
assert f"{shop.shop_code} has been activated" in message
|
|
|
|
def test_toggle_shop_status_not_found(self, db):
|
|
"""Test toggle shop status when shop not found"""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
self.service.toggle_shop_status(db, 99999)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "Shop not found" in str(exc_info.value.detail)
|
|
|
|
def test_get_marketplace_import_jobs_no_filters(self, db, test_marketplace_job):
|
|
"""Test getting marketplace import jobs without filters using fixture"""
|
|
result = self.service.get_marketplace_import_jobs(db, skip=0, limit=10)
|
|
|
|
assert len(result) >= 1
|
|
# Find our test job in the results
|
|
test_job = next((job for job in result if job.job_id == test_marketplace_job.id), None)
|
|
assert test_job is not None
|
|
assert test_job.marketplace == test_marketplace_job.marketplace
|
|
assert test_job.shop_name == test_marketplace_job.shop_name
|
|
assert test_job.status == test_marketplace_job.status
|
|
|
|
def test_get_marketplace_import_jobs_with_marketplace_filter(self, db, test_marketplace_job):
|
|
"""Test getting marketplace import jobs filtered by marketplace"""
|
|
# Create additional job with different marketplace
|
|
other_job = MarketplaceImportJob(
|
|
marketplace="ebay",
|
|
shop_name="eBay Shop",
|
|
status="completed",
|
|
source_url="https://ebay.example.com/import"
|
|
)
|
|
db.add(other_job)
|
|
db.commit()
|
|
|
|
# Filter by the test marketplace job's marketplace
|
|
result = self.service.get_marketplace_import_jobs(db, marketplace=test_marketplace_job.marketplace)
|
|
|
|
assert len(result) >= 1
|
|
# All results should match the marketplace filter
|
|
for job in result:
|
|
assert test_marketplace_job.marketplace.lower() in job.marketplace.lower()
|
|
|
|
def test_get_marketplace_import_jobs_with_shop_filter(self, db, test_marketplace_job):
|
|
"""Test getting marketplace import jobs filtered by shop name"""
|
|
# Create additional job with different shop name
|
|
other_job = MarketplaceImportJob(
|
|
marketplace="amazon",
|
|
shop_name="Different Shop Name",
|
|
status="completed",
|
|
source_url="https://different.example.com/import"
|
|
)
|
|
db.add(other_job)
|
|
db.commit()
|
|
|
|
# Filter by the test marketplace job's shop name
|
|
result = self.service.get_marketplace_import_jobs(db, shop_name=test_marketplace_job.shop_name)
|
|
|
|
assert len(result) >= 1
|
|
# All results should match the shop name filter
|
|
for job in result:
|
|
assert test_marketplace_job.shop_name.lower() in job.shop_name.lower()
|
|
|
|
def test_get_marketplace_import_jobs_with_status_filter(self, db, test_marketplace_job):
|
|
"""Test getting marketplace import jobs filtered by status"""
|
|
# Create additional job with different status
|
|
other_job = MarketplaceImportJob(
|
|
marketplace="amazon",
|
|
shop_name="Test Shop",
|
|
status="pending",
|
|
source_url="https://pending.example.com/import"
|
|
)
|
|
db.add(other_job)
|
|
db.commit()
|
|
|
|
# Filter by the test marketplace job's status
|
|
result = self.service.get_marketplace_import_jobs(db, status=test_marketplace_job.status)
|
|
|
|
assert len(result) >= 1
|
|
# All results should match the status filter
|
|
for job in result:
|
|
assert job.status == test_marketplace_job.status
|
|
|
|
def test_get_marketplace_import_jobs_with_multiple_filters(self, db, test_marketplace_job, test_shop):
|
|
"""Test getting marketplace import jobs with multiple filters"""
|
|
# Create jobs that don't match all filters
|
|
non_matching_job1 = MarketplaceImportJob(
|
|
marketplace="ebay", # Different marketplace
|
|
shop_name=test_marketplace_job.shop_name,
|
|
status=test_marketplace_job.status,
|
|
source_url="https://non-matching1.example.com/import",
|
|
shop_id=test_shop.id # Add required shop_id
|
|
)
|
|
non_matching_job2 = MarketplaceImportJob(
|
|
marketplace=test_marketplace_job.marketplace,
|
|
shop_name="Different Shop", # Different shop name
|
|
status=test_marketplace_job.status,
|
|
source_url="https://non-matching2.example.com/import",
|
|
shop_id=test_shop.id # Add required shop_id
|
|
)
|
|
db.add_all([non_matching_job1, non_matching_job2])
|
|
db.commit()
|
|
|
|
# Apply all three filters matching the test job
|
|
result = self.service.get_marketplace_import_jobs(
|
|
db,
|
|
marketplace=test_marketplace_job.marketplace,
|
|
shop_name=test_marketplace_job.shop_name,
|
|
status=test_marketplace_job.status
|
|
)
|
|
|
|
assert len(result) >= 1
|
|
# Find our test job in the results
|
|
test_job = next((job for job in result if job.job_id == test_marketplace_job.id), None)
|
|
assert test_job is not None
|
|
assert test_job.marketplace == test_marketplace_job.marketplace
|
|
assert test_job.shop_name == test_marketplace_job.shop_name
|
|
assert test_job.status == test_marketplace_job.status
|
|
|
|
def test_get_marketplace_import_jobs_null_values(self, db):
|
|
"""Test that marketplace import jobs handle null values correctly"""
|
|
# Create job with null values but required fields
|
|
job = MarketplaceImportJob(
|
|
marketplace="test",
|
|
shop_name="Test Shop",
|
|
status="pending",
|
|
source_url="https://test.example.com/import",
|
|
imported_count=None,
|
|
updated_count=None,
|
|
total_processed=None,
|
|
error_count=None,
|
|
error_message=None
|
|
)
|
|
db.add(job)
|
|
db.commit()
|
|
|
|
result = self.service.get_marketplace_import_jobs(db)
|
|
|
|
assert len(result) >= 1
|
|
# Find the job with null values
|
|
null_job = next((j for j in result if j.job_id == job.id), None)
|
|
assert null_job is not None
|
|
assert null_job.imported == 0 # None converted to 0
|
|
assert null_job.updated == 0
|
|
assert null_job.total_processed == 0
|
|
assert null_job.error_count == 0
|
|
assert null_job.error_message is None
|
|
|
|
def test_get_user_by_id(self, db, test_user):
|
|
"""Test getting user by ID using fixture"""
|
|
user = self.service.get_user_by_id(db, test_user.id)
|
|
|
|
assert user is not None
|
|
assert user.id == test_user.id
|
|
assert user.email == test_user.email
|
|
assert user.username == test_user.username
|
|
|
|
def test_get_user_by_id_not_found(self, db):
|
|
"""Test getting user by ID when user doesn't exist"""
|
|
user = self.service.get_user_by_id(db, 99999)
|
|
|
|
assert user is None
|
|
|
|
def test_get_shop_by_id(self, db, test_shop):
|
|
"""Test getting shop by ID using fixture"""
|
|
shop = self.service.get_shop_by_id(db, test_shop.id)
|
|
|
|
assert shop is not None
|
|
assert shop.id == test_shop.id
|
|
assert shop.shop_code == test_shop.shop_code
|
|
assert shop.shop_name == test_shop.shop_name
|
|
|
|
def test_get_shop_by_id_not_found(self, db):
|
|
"""Test getting shop by ID when shop doesn't exist"""
|
|
shop = self.service.get_shop_by_id(db, 99999)
|
|
|
|
assert shop is None
|
|
|
|
def test_user_exists_true(self, db, test_user):
|
|
"""Test user_exists returns True when user exists"""
|
|
exists = self.service.user_exists(db, test_user.id)
|
|
|
|
assert exists is True
|
|
|
|
def test_user_exists_false(self, db):
|
|
"""Test user_exists returns False when user doesn't exist"""
|
|
exists = self.service.user_exists(db, 99999)
|
|
|
|
assert exists is False
|
|
|
|
def test_shop_exists_true(self, db, test_shop):
|
|
"""Test shop_exists returns True when shop exists"""
|
|
exists = self.service.shop_exists(db, test_shop.id)
|
|
|
|
assert exists is True
|
|
|
|
def test_shop_exists_false(self, db):
|
|
"""Test shop_exists returns False when shop doesn't exist"""
|
|
exists = self.service.shop_exists(db, 99999)
|
|
|
|
assert exists is False
|