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

code quality run

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-13 21:58:54 +02:00
parent 0dfd885847
commit 3eb18ef91e
63 changed files with 1802 additions and 1289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
tests/test_marketplace_service.py
View File

@@ -1,10 +1,12 @@
# tests/test_marketplace_service.py
import pytest
import uuid
from datetime import datetime
import pytest
from app.services.marketplace_service import MarketplaceService
from models.api_models import MarketplaceImportRequest
from models.database_models import MarketplaceImportJob, Shop, User
from datetime import datetime
class TestMarketplaceService:
@@ -22,7 +24,9 @@ class TestMarketplaceService:
assert result.shop_code == test_shop.shop_code
assert result.owner_id == test_user.id
def test_validate_shop_access_admin_can_access_any_shop(self, db, test_shop, test_admin):
def test_validate_shop_access_admin_can_access_any_shop(
self, db, test_shop, test_admin
):
"""Test that admin users can access any shop"""
result = self.service.validate_shop_access(db, test_shop.shop_code, test_admin)
@@ -33,7 +37,9 @@ class TestMarketplaceService:
with pytest.raises(ValueError, match="Shop not found"):
self.service.validate_shop_access(db, "NONEXISTENT", test_user)
def test_validate_shop_access_permission_denied(self, db, test_shop, test_user, other_user):
def test_validate_shop_access_permission_denied(
self, db, test_shop, test_user, other_user
):
"""Test shop access validation when user doesn't own the shop"""
# Set the shop owner to a different user
test_shop.owner_id = other_user.id
@@ -52,7 +58,7 @@ class TestMarketplaceService:
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code=test_shop.shop_code,
batch_size=1000
batch_size=1000,
)
result = self.service.create_import_job(db, request, test_user)
@@ -60,7 +66,7 @@ class TestMarketplaceService:
assert result.marketplace == "Amazon"
# Check the correct field based on your model
assert result.shop_id == test_shop.id # Changed from shop_code to shop_id
assert result.user_id == test_user.id if hasattr(result, 'user_id') else True
assert result.user_id == test_user.id if hasattr(result, "user_id") else True
assert result.status == "pending"
assert result.source_url == "https://example.com/products.csv"
@@ -70,7 +76,7 @@ class TestMarketplaceService:
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code="INVALID_SHOP",
batch_size=1000
batch_size=1000,
)
with pytest.raises(ValueError, match="Shop not found"):
@@ -78,16 +84,22 @@ class TestMarketplaceService:
def test_get_import_job_by_id_success(self, db, test_marketplace_job, test_user):
"""Test getting import job by ID for job owner"""
result = self.service.get_import_job_by_id(db, test_marketplace_job.id, test_user)
result = self.service.get_import_job_by_id(
db, test_marketplace_job.id, test_user
)
assert result.id == test_marketplace_job.id
# Check user_id if the field exists
if hasattr(result, 'user_id'):
if hasattr(result, "user_id"):
assert result.user_id == test_user.id
def test_get_import_job_by_id_admin_access(self, db, test_marketplace_job, test_admin):
def test_get_import_job_by_id_admin_access(
self, db, test_marketplace_job, test_admin
):
"""Test that admin can access any import job"""
result = self.service.get_import_job_by_id(db, test_marketplace_job.id, test_admin)
result = self.service.get_import_job_by_id(
db, test_marketplace_job.id, test_admin
)
assert result.id == test_marketplace_job.id
@@ -96,7 +108,9 @@ class TestMarketplaceService:
with pytest.raises(ValueError, match="Marketplace import job not found"):
self.service.get_import_job_by_id(db, 99999, test_user)
def test_get_import_job_by_id_access_denied(self, db, test_marketplace_job, other_user):
def test_get_import_job_by_id_access_denied(
self, db, test_marketplace_job, other_user
):
"""Test access denied when user doesn't own the job"""
with pytest.raises(PermissionError, match="Access denied to this import job"):
self.service.get_import_job_by_id(db, test_marketplace_job.id, other_user)
@@ -108,7 +122,7 @@ class TestMarketplaceService:
assert len(jobs) >= 1
assert any(job.id == test_marketplace_job.id for job in jobs)
# Check user_id if the field exists
if hasattr(test_marketplace_job, 'user_id'):
if hasattr(test_marketplace_job, "user_id"):
assert test_marketplace_job.user_id == test_user.id
def test_get_import_jobs_admin_sees_all(self, db, test_marketplace_job, test_admin):
@@ -118,7 +132,9 @@ class TestMarketplaceService:
assert len(jobs) >= 1
assert any(job.id == test_marketplace_job.id for job in jobs)
def test_get_import_jobs_with_marketplace_filter(self, db, test_marketplace_job, test_user):
def test_get_import_jobs_with_marketplace_filter(
self, db, test_marketplace_job, test_user
):
"""Test getting import jobs with marketplace filter"""
jobs = self.service.get_import_jobs(
db, test_user, marketplace=test_marketplace_job.marketplace
@@ -143,7 +159,7 @@ class TestMarketplaceService:
imported_count=0,
updated_count=0,
total_processed=0,
error_count=0
error_count=0,
)
db.add(job)
db.commit()
@@ -159,7 +175,7 @@ class TestMarketplaceService:
test_marketplace_job.id,
"completed",
imported_count=100,
total_processed=100
total_processed=100,
)
assert result.status == "completed"
@@ -211,7 +227,7 @@ class TestMarketplaceService:
imported_count=0,
updated_count=0,
total_processed=0,
error_count=0
error_count=0,
)
db.add(job)
db.commit()
@@ -222,13 +238,17 @@ class TestMarketplaceService:
assert result.status == "cancelled"
assert result.completed_at is not None
def test_cancel_import_job_invalid_status(self, db, test_marketplace_job, test_user):
def test_cancel_import_job_invalid_status(
self, db, test_marketplace_job, test_user
):
"""Test cancelling a job that can't be cancelled"""
# Set job status to completed
test_marketplace_job.status = "completed"
db.commit()
with pytest.raises(ValueError, match="Cannot cancel job with status: completed"):
with pytest.raises(
ValueError, match="Cannot cancel job with status: completed"
):
self.service.cancel_import_job(db, test_marketplace_job.id, test_user)
def test_delete_import_job_success(self, db, test_user, test_shop):
@@ -246,7 +266,7 @@ class TestMarketplaceService:
imported_count=0,
updated_count=0,
total_processed=0,
error_count=0
error_count=0,
)
db.add(job)
db.commit()
@@ -258,7 +278,11 @@ class TestMarketplaceService:
assert result is True
# Verify the job is actually deleted
deleted_job = db.query(MarketplaceImportJob).filter(MarketplaceImportJob.id == job_id).first()
deleted_job = (
db.query(MarketplaceImportJob)
.filter(MarketplaceImportJob.id == job_id)
.first()
)
assert deleted_job is None
def test_delete_import_job_invalid_status(self, db, test_user, test_shop):
@@ -276,7 +300,7 @@ class TestMarketplaceService:
imported_count=0,
updated_count=0,
total_processed=0,
error_count=0
error_count=0,
)
db.add(job)
db.commit()