diff --git a/tests/integration/api/v1/admin/test_auth.py b/tests/integration/api/v1/admin/test_auth.py index 9163bcb9..2d76207c 100644 --- a/tests/integration/api/v1/admin/test_auth.py +++ b/tests/integration/api/v1/admin/test_auth.py @@ -175,3 +175,48 @@ class TestAdminAuthAPI: assert response.status_code == 200 data = response.json() assert data["message"] == "Logged out successfully" + + def test_super_admin_login_includes_is_super_admin( + self, client, test_super_admin + ): + """Test super admin login includes is_super_admin in response.""" + response = client.post( + "/api/v1/admin/auth/login", + json={ + "email_or_username": test_super_admin.username, + "password": "superadminpass123", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert "user" in data + assert "is_super_admin" in data["user"] + assert data["user"]["is_super_admin"] is True + + def test_platform_admin_login_includes_is_super_admin( + self, client, test_platform_admin + ): + """Test platform admin login includes is_super_admin=False in response.""" + response = client.post( + "/api/v1/admin/auth/login", + json={ + "email_or_username": test_platform_admin.username, + "password": "platformadminpass123", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert "user" in data + assert "is_super_admin" in data["user"] + assert data["user"]["is_super_admin"] is False + + def test_get_current_super_admin_info(self, client, super_admin_headers, test_super_admin): + """Test getting current super admin user info includes is_super_admin.""" + response = client.get("/api/v1/admin/auth/me", headers=super_admin_headers) + + assert response.status_code == 200 + data = response.json() + assert data["username"] == test_super_admin.username + assert data["is_super_admin"] is True diff --git a/tests/integration/api/v1/admin/test_vendors.py b/tests/integration/api/v1/admin/test_vendors.py index 3b180708..577b6922 100644 --- a/tests/integration/api/v1/admin/test_vendors.py +++ b/tests/integration/api/v1/admin/test_vendors.py @@ -115,3 +115,95 @@ class TestAdminVendorsAPI: assert len(data["vendors"]) >= 0 assert "skip" in data assert "limit" in data + + +@pytest.mark.integration +@pytest.mark.api +@pytest.mark.admin +class TestAdminVendorCreationAPI: + """Test admin vendor creation endpoints with platform assignment.""" + + def test_create_vendor_without_platforms( + self, client, admin_headers, test_company + ): + """Test creating a vendor without platform assignments.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + response = client.post( + "/api/v1/admin/vendors", + headers=admin_headers, + json={ + "company_id": test_company.id, + "vendor_code": f"NOPLAT_{unique_id}", + "subdomain": f"noplat{unique_id}", + "name": f"No Platform Vendor {unique_id}", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["vendor_code"] == f"NOPLAT_{unique_id}".upper() + assert data["company_id"] == test_company.id + + def test_create_vendor_with_platforms( + self, client, admin_headers, test_company, test_platform, another_platform + ): + """Test creating a vendor with platform assignments.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + response = client.post( + "/api/v1/admin/vendors", + headers=admin_headers, + json={ + "company_id": test_company.id, + "vendor_code": f"WITHPLAT_{unique_id}", + "subdomain": f"withplat{unique_id}", + "name": f"With Platform Vendor {unique_id}", + "platform_ids": [test_platform.id, another_platform.id], + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["vendor_code"] == f"WITHPLAT_{unique_id}".upper() + + def test_create_vendor_duplicate_code_fails( + self, client, admin_headers, test_company, test_vendor + ): + """Test creating a vendor with duplicate code fails.""" + response = client.post( + "/api/v1/admin/vendors", + headers=admin_headers, + json={ + "company_id": test_company.id, + "vendor_code": test_vendor.vendor_code, + "subdomain": "uniquesubdomain123", + "name": "Duplicate Code Vendor", + }, + ) + + assert response.status_code == 409 # Conflict + data = response.json() + assert data["error_code"] == "VENDOR_ALREADY_EXISTS" + + def test_create_vendor_non_admin_fails( + self, client, auth_headers, test_company + ): + """Test non-admin cannot create vendors.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + response = client.post( + "/api/v1/admin/vendors", + headers=auth_headers, + json={ + "company_id": test_company.id, + "vendor_code": f"NONADMIN_{unique_id}", + "subdomain": f"nonadmin{unique_id}", + "name": "Non Admin Vendor", + }, + ) + + assert response.status_code == 403 diff --git a/tests/unit/services/test_admin_service.py b/tests/unit/services/test_admin_service.py index 2e3d2d1b..c946c785 100644 --- a/tests/unit/services/test_admin_service.py +++ b/tests/unit/services/test_admin_service.py @@ -6,10 +6,13 @@ from app.exceptions import ( CannotModifySelfException, UserNotFoundException, UserStatusChangeException, + ValidationException, + VendorAlreadyExistsException, VendorNotFoundException, ) from app.services.admin_service import AdminService from app.services.stats_service import stats_service +from models.schema.vendor import VendorCreate @pytest.mark.unit @@ -316,3 +319,180 @@ class TestAdminService: assert stats["active_vendors"] == 0 assert stats["verified_vendors"] == 0 assert stats["verification_rate"] == 0 + + +@pytest.mark.unit +@pytest.mark.admin +class TestAdminServiceVendorCreation: + """Test suite for AdminService.create_vendor with platform assignments.""" + + def setup_method(self): + """Setup method following the same pattern as other tests.""" + self.service = AdminService() + + def test_create_vendor_without_platforms(self, db, test_company): + """Test creating a vendor without platform assignments.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + vendor_data = VendorCreate( + company_id=test_company.id, + vendor_code=f"NOPLATFORM_{unique_id}", + subdomain=f"noplatform{unique_id}", + name=f"No Platform Vendor {unique_id}", + ) + + vendor = self.service.create_vendor(db, vendor_data) + db.commit() + + assert vendor is not None + assert vendor.vendor_code == vendor_data.vendor_code.upper() + assert vendor.company_id == test_company.id + assert vendor.is_active is True + + def test_create_vendor_with_single_platform( + self, db, test_company, test_platform + ): + """Test creating a vendor with one platform assignment.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + vendor_data = VendorCreate( + company_id=test_company.id, + vendor_code=f"SINGLEPLAT_{unique_id}", + subdomain=f"singleplat{unique_id}", + name=f"Single Platform Vendor {unique_id}", + platform_ids=[test_platform.id], + ) + + vendor = self.service.create_vendor(db, vendor_data) + db.commit() + db.refresh(vendor) + + assert vendor is not None + assert vendor.vendor_code == vendor_data.vendor_code.upper() + + # Verify platform assignment + from models.database.vendor_platform import VendorPlatform + + assignment = ( + db.query(VendorPlatform) + .filter( + VendorPlatform.vendor_id == vendor.id, + VendorPlatform.platform_id == test_platform.id, + ) + .first() + ) + assert assignment is not None + assert assignment.is_active is True + + def test_create_vendor_with_multiple_platforms( + self, db, test_company, test_platform, another_platform + ): + """Test creating a vendor with multiple platform assignments.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + vendor_data = VendorCreate( + company_id=test_company.id, + vendor_code=f"MULTIPLAT_{unique_id}", + subdomain=f"multiplat{unique_id}", + name=f"Multi Platform Vendor {unique_id}", + platform_ids=[test_platform.id, another_platform.id], + ) + + vendor = self.service.create_vendor(db, vendor_data) + db.commit() + db.refresh(vendor) + + assert vendor is not None + + # Verify both platform assignments + from models.database.vendor_platform import VendorPlatform + + assignments = ( + db.query(VendorPlatform) + .filter(VendorPlatform.vendor_id == vendor.id) + .all() + ) + assert len(assignments) == 2 + + platform_ids = [a.platform_id for a in assignments] + assert test_platform.id in platform_ids + assert another_platform.id in platform_ids + + def test_create_vendor_with_invalid_platform_id(self, db, test_company): + """Test creating a vendor with non-existent platform ID (should ignore).""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + vendor_data = VendorCreate( + company_id=test_company.id, + vendor_code=f"INVALIDPLAT_{unique_id}", + subdomain=f"invalidplat{unique_id}", + name=f"Invalid Platform Vendor {unique_id}", + platform_ids=[99999], # Non-existent platform + ) + + # Should succeed but not create assignment for invalid platform + vendor = self.service.create_vendor(db, vendor_data) + db.commit() + db.refresh(vendor) + + assert vendor is not None + + # Verify no platform assignments created + from models.database.vendor_platform import VendorPlatform + + assignments = ( + db.query(VendorPlatform) + .filter(VendorPlatform.vendor_id == vendor.id) + .all() + ) + assert len(assignments) == 0 + + def test_create_vendor_duplicate_code_fails(self, db, test_company, test_vendor): + """Test creating a vendor with duplicate vendor code fails.""" + vendor_data = VendorCreate( + company_id=test_company.id, + vendor_code=test_vendor.vendor_code, # Duplicate + subdomain="uniquesubdomain", + name="Duplicate Code Vendor", + ) + + with pytest.raises(VendorAlreadyExistsException): + self.service.create_vendor(db, vendor_data) + + def test_create_vendor_duplicate_subdomain_fails(self, db, test_company, test_vendor): + """Test creating a vendor with duplicate subdomain fails.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + vendor_data = VendorCreate( + company_id=test_company.id, + vendor_code=f"UNIQUECODE_{unique_id}", + subdomain=test_vendor.subdomain, # Duplicate + name="Duplicate Subdomain Vendor", + ) + + with pytest.raises(ValidationException) as exc_info: + self.service.create_vendor(db, vendor_data) + + assert "already taken" in str(exc_info.value) + + def test_create_vendor_invalid_company_fails(self, db): + """Test creating a vendor with non-existent company fails.""" + import uuid + + unique_id = str(uuid.uuid4())[:8] + vendor_data = VendorCreate( + company_id=99999, # Non-existent + vendor_code=f"NOCOMPANY_{unique_id}", + subdomain=f"nocompany{unique_id}", + name="No Company Vendor", + ) + + with pytest.raises(ValidationException) as exc_info: + self.service.create_vendor(db, vendor_data) + + assert "not found" in str(exc_info.value)