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

Application fully migrated to modular approach

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-13 21:30:40 +02:00
parent c7d6b33cd5
commit b9fe91ab88
38 changed files with 509 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
tests/test_marketplace_service.py
View File

@@ -76,56 +76,56 @@ class TestMarketplaceService:
with pytest.raises(ValueError, match="Shop not found"):
self.service.create_import_job(db, request, test_user)
def test_get_import_job_by_id_success(self, db, test_import_job, test_user):
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_import_job.id, test_user)
result = self.service.get_import_job_by_id(db, test_marketplace_job.id, test_user)
assert result.id == test_import_job.id
assert result.id == test_marketplace_job.id
# Check user_id if the field exists
if hasattr(result, 'user_id'):
assert result.user_id == test_user.id
def test_get_import_job_by_id_admin_access(self, db, test_import_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_import_job.id, test_admin)
result = self.service.get_import_job_by_id(db, test_marketplace_job.id, test_admin)
assert result.id == test_import_job.id
assert result.id == test_marketplace_job.id
def test_get_import_job_by_id_not_found(self, db, test_user):
"""Test getting non-existent import job"""
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_import_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_import_job.id, other_user)
self.service.get_import_job_by_id(db, test_marketplace_job.id, other_user)
def test_get_import_jobs_user_filter(self, db, test_import_job, test_user):
def test_get_import_jobs_user_filter(self, db, test_marketplace_job, test_user):
"""Test getting import jobs filtered by user"""
jobs = self.service.get_import_jobs(db, test_user)
assert len(jobs) >= 1
assert any(job.id == test_import_job.id for job in jobs)
assert any(job.id == test_marketplace_job.id for job in jobs)
# Check user_id if the field exists
if hasattr(test_import_job, 'user_id'):
assert test_import_job.user_id == test_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_import_job, test_admin):
def test_get_import_jobs_admin_sees_all(self, db, test_marketplace_job, test_admin):
"""Test that admin sees all import jobs"""
jobs = self.service.get_import_jobs(db, test_admin)
assert len(jobs) >= 1
assert any(job.id == test_import_job.id for job in jobs)
assert any(job.id == test_marketplace_job.id for job in jobs)
def test_get_import_jobs_with_marketplace_filter(self, db, test_import_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_import_job.marketplace
db, test_user, marketplace=test_marketplace_job.marketplace
)
assert len(jobs) >= 1
assert any(job.marketplace == test_import_job.marketplace for job in jobs)
assert any(job.marketplace == test_marketplace_job.marketplace for job in jobs)
def test_get_import_jobs_with_pagination(self, db, test_user, test_shop):
"""Test getting import jobs with pagination"""
@@ -137,6 +137,7 @@ class TestMarketplaceService:
status="completed",
marketplace=f"Marketplace_{unique_id}_{i}",
shop_name=f"Test_Shop_{unique_id}_{i}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
source_url=f"https://test-{i}.example.com/import",
imported_count=0,
@@ -151,11 +152,11 @@ class TestMarketplaceService:
assert len(jobs) <= 2 # Should be at most 2
def test_update_job_status_success(self, db, test_import_job):
def test_update_job_status_success(self, db, test_marketplace_job):
"""Test updating job status"""
result = self.service.update_job_status(
db,
test_import_job.id,
test_marketplace_job.id,
"completed",
imported_count=100,
total_processed=100
@@ -170,7 +171,7 @@ class TestMarketplaceService:
with pytest.raises(ValueError, match="Marketplace import job not found"):
self.service.update_job_status(db, 99999, "completed")
def test_get_job_stats_user(self, db, test_import_job, test_user):
def test_get_job_stats_user(self, db, test_marketplace_job, test_user):
"""Test getting job statistics for user"""
stats = self.service.get_job_stats(db, test_user)
@@ -180,20 +181,20 @@ class TestMarketplaceService:
assert "completed_jobs" in stats
assert "failed_jobs" in stats
def test_get_job_stats_admin(self, db, test_import_job, test_admin):
def test_get_job_stats_admin(self, db, test_marketplace_job, test_admin):
"""Test getting job statistics for admin"""
stats = self.service.get_job_stats(db, test_admin)
assert stats["total_jobs"] >= 1
def test_convert_to_response_model(self, test_import_job):
def test_convert_to_response_model(self, test_marketplace_job):
"""Test converting database model to response model"""
response = self.service.convert_to_response_model(test_import_job)
response = self.service.convert_to_response_model(test_marketplace_job)
assert response.job_id == test_import_job.id
assert response.status == test_import_job.status
assert response.marketplace == test_import_job.marketplace
assert response.imported == (test_import_job.imported_count or 0)
assert response.job_id == test_marketplace_job.id
assert response.status == test_marketplace_job.status
assert response.marketplace == test_marketplace_job.marketplace
assert response.imported == (test_marketplace_job.imported_count or 0)
def test_cancel_import_job_success(self, db, test_user, test_shop):
"""Test cancelling a pending import job"""
@@ -204,6 +205,7 @@ class TestMarketplaceService:
status="pending",
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
source_url="https://test.example.com/import",
imported_count=0,
@@ -220,14 +222,14 @@ class TestMarketplaceService:
assert result.status == "cancelled"
assert result.completed_at is not None
def test_cancel_import_job_invalid_status(self, db, test_import_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_import_job.status = "completed"
test_marketplace_job.status = "completed"
db.commit()
with pytest.raises(ValueError, match="Cannot cancel job with status: completed"):
self.service.cancel_import_job(db, test_import_job.id, test_user)
self.service.cancel_import_job(db, test_marketplace_job.id, test_user)
def test_delete_import_job_success(self, db, test_user, test_shop):
"""Test deleting a completed import job"""
@@ -238,6 +240,7 @@ class TestMarketplaceService:
status="completed",
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
source_url="https://test.example.com/import",
imported_count=0,
@@ -267,6 +270,7 @@ class TestMarketplaceService:
status="pending",
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
source_url="https://test.example.com/import",
imported_count=0,