diff --git a/middleware/auth.py b/middleware/auth.py index 5f85c13e..5b4589b7 100644 --- a/middleware/auth.py +++ b/middleware/auth.py @@ -100,7 +100,7 @@ class AuthManager: if exp is None: raise InvalidTokenException("Token missing expiration") - if datetime.now(timezone.utc) > datetime.fromtimestamp(exp): + if datetime.now(timezone.utc) > datetime.fromtimestamp(exp, tz=timezone.utc): raise TokenExpiredException() # Extract user data diff --git a/tests/integration/api/v1/test_export.py b/tests/integration/api/v1/test_product_export.py similarity index 100% rename from tests/integration/api/v1/test_export.py rename to tests/integration/api/v1/test_product_export.py diff --git a/tests/unit/services/test_product_service.py b/tests/unit/services/test_product_service.py index d94a755d..72f1b8dc 100644 --- a/tests/unit/services/test_product_service.py +++ b/tests/unit/services/test_product_service.py @@ -2,7 +2,14 @@ import pytest from app.services.product_service import ProductService -from models.schemas.product import ProductCreate +from app.exceptions import ( + ProductNotFoundException, + ProductAlreadyExistsException, + InvalidProductDataException, + ProductValidationException, + ValidationException, +) +from models.schemas.product import ProductCreate, ProductUpdate from models.database.product import Product @@ -12,8 +19,8 @@ class TestProductService: def setup_method(self): self.service = ProductService() - def test_create_product_with_gtin_validation(self, db): - """Test product creation with GTIN validation""" + def test_create_product_success(self, db): + """Test successful product creation with valid data""" product_data = ProductCreate( product_id="SVC001", title="Service Test Product", @@ -25,11 +32,13 @@ class TestProductService: product = self.service.create_product(db, product_data) assert product.product_id == "SVC001" + assert product.title == "Service Test Product" assert product.gtin == "1234567890123" assert product.marketplace == "TestMarket" + assert product.price == "19.99" # Price is stored as string after processing def test_create_product_invalid_gtin(self, db): - """Test product creation with invalid GTIN""" + """Test product creation with invalid GTIN raises InvalidProductDataException""" product_data = ProductCreate( product_id="SVC002", title="Service Test Product", @@ -37,22 +46,238 @@ class TestProductService: price="19.99", ) - with pytest.raises(ValueError, match="Invalid GTIN format"): + with pytest.raises(InvalidProductDataException) as exc_info: self.service.create_product(db, product_data) - def test_get_products_with_filters(self, db, test_product): + assert exc_info.value.error_code == "INVALID_PRODUCT_DATA" + assert "Invalid GTIN format" in str(exc_info.value) + assert exc_info.value.status_code == 422 + assert exc_info.value.details.get("field") == "gtin" + + def test_create_product_missing_product_id(self, db): + """Test product creation without product_id raises ProductValidationException""" + product_data = ProductCreate( + product_id="", # Empty product ID + title="Service Test Product", + price="19.99", + ) + + with pytest.raises(ProductValidationException) as exc_info: + self.service.create_product(db, product_data) + + assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED" + assert "Product ID is required" in str(exc_info.value) + assert exc_info.value.details.get("field") == "product_id" + + def test_create_product_missing_title(self, db): + """Test product creation without title raises ProductValidationException""" + product_data = ProductCreate( + product_id="SVC003", + title="", # Empty title + price="19.99", + ) + + with pytest.raises(ProductValidationException) as exc_info: + self.service.create_product(db, product_data) + + assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED" + assert "Product title is required" in str(exc_info.value) + assert exc_info.value.details.get("field") == "title" + + def test_create_product_already_exists(self, db, test_product): + """Test creating product with existing ID raises ProductAlreadyExistsException""" + product_data = ProductCreate( + product_id=test_product.product_id, # Use existing product ID + title="Duplicate Product", + price="29.99", + ) + + with pytest.raises(ProductAlreadyExistsException) as exc_info: + self.service.create_product(db, product_data) + + assert exc_info.value.error_code == "PRODUCT_ALREADY_EXISTS" + assert test_product.product_id in str(exc_info.value) + assert exc_info.value.status_code == 409 + assert exc_info.value.details.get("product_id") == test_product.product_id + + def test_create_product_invalid_price(self, db): + """Test product creation with invalid price raises InvalidProductDataException""" + product_data = ProductCreate( + product_id="SVC004", + title="Service Test Product", + price="invalid_price", + ) + + with pytest.raises(InvalidProductDataException) as exc_info: + self.service.create_product(db, product_data) + + assert exc_info.value.error_code == "INVALID_PRODUCT_DATA" + assert "Invalid price format" in str(exc_info.value) + assert exc_info.value.details.get("field") == "price" + + def test_get_product_by_id_or_raise_success(self, db, test_product): + """Test successful product retrieval by ID""" + product = self.service.get_product_by_id_or_raise(db, test_product.product_id) + + assert product.product_id == test_product.product_id + assert product.title == test_product.title + + def test_get_product_by_id_or_raise_not_found(self, db): + """Test product retrieval with non-existent ID raises ProductNotFoundException""" + with pytest.raises(ProductNotFoundException) as exc_info: + self.service.get_product_by_id_or_raise(db, "NONEXISTENT") + + assert exc_info.value.error_code == "PRODUCT_NOT_FOUND" + assert "NONEXISTENT" in str(exc_info.value) + assert exc_info.value.status_code == 404 + assert exc_info.value.details.get("resource_type") == "Product" + assert exc_info.value.details.get("identifier") == "NONEXISTENT" + + def test_get_products_with_filters_success(self, db, test_product): """Test getting products with various filters""" - products, total = self.service.get_products_with_filters(db, brand="TestBrand") - - assert total == 1 - assert len(products) == 1 - assert products[0].brand == "TestBrand" - - def test_get_products_with_search(self, db, test_product): - """Test getting products with search""" products, total = self.service.get_products_with_filters( - db, search="Test Product" + db, brand=test_product.brand ) assert total == 1 assert len(products) == 1 + assert products[0].brand == test_product.brand + + def test_get_products_with_search(self, db, test_product): + """Test getting products with search term""" + products, total = self.service.get_products_with_filters( + db, search="Test Product" + ) + + assert total >= 1 + assert len(products) >= 1 + # Verify search worked by checking that title contains search term + found_product = next((p for p in products if p.product_id == test_product.product_id), None) + assert found_product is not None + + def test_update_product_success(self, db, test_product): + """Test successful product update""" + update_data = ProductUpdate( + title="Updated Product Title", + price="39.99" + ) + + updated_product = self.service.update_product(db, test_product.product_id, update_data) + + assert updated_product.title == "Updated Product Title" + assert updated_product.price == "39.99" # Price is stored as string after processing + assert updated_product.product_id == test_product.product_id # ID unchanged + + def test_update_product_not_found(self, db): + """Test updating non-existent product raises ProductNotFoundException""" + update_data = ProductUpdate(title="Updated Title") + + with pytest.raises(ProductNotFoundException) as exc_info: + self.service.update_product(db, "NONEXISTENT", update_data) + + assert exc_info.value.error_code == "PRODUCT_NOT_FOUND" + assert "NONEXISTENT" in str(exc_info.value) + + def test_update_product_invalid_gtin(self, db, test_product): + """Test updating product with invalid GTIN raises InvalidProductDataException""" + update_data = ProductUpdate(gtin="invalid_gtin") + + with pytest.raises(InvalidProductDataException) as exc_info: + self.service.update_product(db, test_product.product_id, update_data) + + assert exc_info.value.error_code == "INVALID_PRODUCT_DATA" + assert "Invalid GTIN format" in str(exc_info.value) + assert exc_info.value.details.get("field") == "gtin" + + def test_update_product_empty_title(self, db, test_product): + """Test updating product with empty title raises ProductValidationException""" + update_data = ProductUpdate(title="") + + with pytest.raises(ProductValidationException) as exc_info: + self.service.update_product(db, test_product.product_id, update_data) + + assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED" + assert "Product title cannot be empty" in str(exc_info.value) + assert exc_info.value.details.get("field") == "title" + + def test_update_product_invalid_price(self, db, test_product): + """Test updating product with invalid price raises InvalidProductDataException""" + update_data = ProductUpdate(price="invalid_price") + + with pytest.raises(InvalidProductDataException) as exc_info: + self.service.update_product(db, test_product.product_id, update_data) + + assert exc_info.value.error_code == "INVALID_PRODUCT_DATA" + assert "Invalid price format" in str(exc_info.value) + assert exc_info.value.details.get("field") == "price" + + def test_delete_product_success(self, db, test_product): + """Test successful product deletion""" + result = self.service.delete_product(db, test_product.product_id) + + assert result is True + + # Verify product is deleted + deleted_product = self.service.get_product_by_id(db, test_product.product_id) + assert deleted_product is None + + def test_delete_product_not_found(self, db): + """Test deleting non-existent product raises ProductNotFoundException""" + with pytest.raises(ProductNotFoundException) as exc_info: + self.service.delete_product(db, "NONEXISTENT") + + assert exc_info.value.error_code == "PRODUCT_NOT_FOUND" + assert "NONEXISTENT" in str(exc_info.value) + + def test_get_stock_info_success(self, db, test_product_with_stock): + """Test getting stock info for product with stock""" + stock_info = self.service.get_stock_info(db, test_product_with_stock.gtin) + + assert stock_info is not None + assert stock_info.gtin == test_product_with_stock.gtin + assert stock_info.total_quantity > 0 + assert len(stock_info.locations) > 0 + + def test_get_stock_info_no_stock(self, db, test_product): + """Test getting stock info for product without stock""" + stock_info = self.service.get_stock_info(db, test_product.gtin or "1234567890123") + + assert stock_info is None + + def test_product_exists_true(self, db, test_product): + """Test product_exists returns True for existing product""" + exists = self.service.product_exists(db, test_product.product_id) + assert exists is True + + def test_product_exists_false(self, db): + """Test product_exists returns False for non-existent product""" + exists = self.service.product_exists(db, "NONEXISTENT") + assert exists is False + + def test_generate_csv_export_success(self, db, test_product): + """Test CSV export generation""" + csv_generator = self.service.generate_csv_export(db) + + # Convert generator to list to test content + csv_lines = list(csv_generator) + + assert len(csv_lines) > 1 # Header + at least one data row + assert csv_lines[0].startswith("product_id,title,description") # Check header + + # Check that test product appears in CSV + csv_content = "".join(csv_lines) + assert test_product.product_id in csv_content + + def test_generate_csv_export_with_filters(self, db, test_product): + """Test CSV export with marketplace filter""" + csv_generator = self.service.generate_csv_export( + db, + marketplace=test_product.marketplace + ) + + csv_lines = list(csv_generator) + assert len(csv_lines) >= 1 # At least header + + if len(csv_lines) > 1: # If there's data + csv_content = "".join(csv_lines) + assert test_product.marketplace in csv_content