From dcc06dee20cdc74e13761846ddd0bbd7867291ed Mon Sep 17 00:00:00 2001 From: Escape0707 Date: Thu, 2 Jul 2026 20:58:56 +0900 Subject: [PATCH] test: migrate tag service tests to testcontainers (#38313) --- api/services/tag_service.py | 27 +- .../pyrefly.toml | 1 - .../services/test_tag_service.py | 2320 +++++++---------- .../unit_tests/services/test_tag_service.py | 172 -- 4 files changed, 1029 insertions(+), 1491 deletions(-) delete mode 100644 api/tests/unit_tests/services/test_tag_service.py diff --git a/api/services/tag_service.py b/api/services/tag_service.py index 45deedb5b93..1827720dfc9 100644 --- a/api/services/tag_service.py +++ b/api/services/tag_service.py @@ -14,6 +14,9 @@ from models.enums import TagType from models.model import App, Tag, TagBinding from models.snippet import CustomizedSnippet +type _SessionLike = Session | scoped_session +type _TagTypeLike = TagType | str + class SaveTagPayload(BaseModel): name: str = Field(min_length=1, max_length=50) @@ -38,7 +41,7 @@ class TagBindingDeletePayload(BaseModel): class TagService: @staticmethod - def get_tags(session: Session, tag_type: str, current_tenant_id: str, keyword: str | None = None): + def get_tags(session: Session, tag_type: _TagTypeLike, current_tenant_id: str, keyword: str | None = None): stmt = ( select(Tag.id, Tag.type, Tag.name, func.count(TagBinding.id).label("binding_count")) .outerjoin(TagBinding, Tag.id == TagBinding.tag_id) @@ -55,10 +58,10 @@ class TagService: @staticmethod def get_target_ids_by_tag_ids( - tag_type: str, + tag_type: _TagTypeLike, current_tenant_id: str, tag_ids: list[str], - session: scoped_session | Session, + session: _SessionLike, *, match_all: bool = False, ): @@ -104,7 +107,7 @@ class TagService: return tag_bindings @staticmethod - def get_tag_by_tag_name(tag_type: str, current_tenant_id: str, tag_name: str, session: scoped_session): + def get_tag_by_tag_name(tag_type: _TagTypeLike, current_tenant_id: str, tag_name: str, session: _SessionLike): if not tag_type or not tag_name: return [] tags = list( @@ -117,7 +120,7 @@ class TagService: return tags @staticmethod - def get_tags_by_target_id(tag_type: str, current_tenant_id: str, target_id: str, session: scoped_session): + def get_tags_by_target_id(tag_type: _TagTypeLike, current_tenant_id: str, target_id: str, session: _SessionLike): tags = session.scalars( select(Tag) .join(TagBinding, Tag.id == TagBinding.tag_id) @@ -132,7 +135,7 @@ class TagService: return tags or [] @staticmethod - def save_tags(payload: SaveTagPayload, session: scoped_session) -> Tag: + def save_tags(payload: SaveTagPayload, session: _SessionLike) -> Tag: if TagService.get_tag_by_tag_name(payload.type, current_user.current_tenant_id, payload.name, session): raise ValueError("Tag name already exists") tag = Tag( @@ -148,7 +151,7 @@ class TagService: @staticmethod def update_tags( - payload: UpdateTagPayload, tag_id: str, session: scoped_session, *, tag_type: TagType | None = None + payload: UpdateTagPayload, tag_id: str, session: _SessionLike, *, tag_type: TagType | None = None ) -> Tag: current_tenant_id = current_user.current_tenant_id stmt = select(Tag).where(Tag.id == tag_id, Tag.tenant_id == current_tenant_id) @@ -175,7 +178,7 @@ class TagService: return tag @staticmethod - def get_tag_binding_count(tag_id: str, session: scoped_session, *, tag_type: TagType | None = None) -> int: + def get_tag_binding_count(tag_id: str, session: _SessionLike, *, tag_type: TagType | None = None) -> int: current_tenant_id = current_user.current_tenant_id stmt = ( select(func.count(TagBinding.id)) @@ -188,7 +191,7 @@ class TagService: return count @staticmethod - def delete_tag(tag_id: str, session: scoped_session, *, tag_type: TagType | None = None): + def delete_tag(tag_id: str, session: _SessionLike, *, tag_type: TagType | None = None): current_tenant_id = current_user.current_tenant_id stmt = select(Tag).where(Tag.id == tag_id, Tag.tenant_id == current_tenant_id) if tag_type is not None: @@ -207,7 +210,7 @@ class TagService: session.commit() @staticmethod - def save_tag_binding(payload: TagBindingCreatePayload, session: scoped_session): + def save_tag_binding(payload: TagBindingCreatePayload, session: _SessionLike): TagService.check_target_exists(payload.type, payload.target_id, session) valid_tag_ids = session.scalars( select(Tag.id).where( @@ -234,7 +237,7 @@ class TagService: session.commit() @staticmethod - def delete_tag_binding(payload: TagBindingDeletePayload, session: scoped_session): + def delete_tag_binding(payload: TagBindingDeletePayload, session: _SessionLike): TagService.check_target_exists(payload.type, payload.target_id, session) result = cast( CursorResult, @@ -257,7 +260,7 @@ class TagService: session.commit() @staticmethod - def check_target_exists(type: str, target_id: str, session: scoped_session): + def check_target_exists(type: _TagTypeLike, target_id: str, session: _SessionLike): if type == "knowledge": dataset = session.scalar( select(Dataset) diff --git a/api/tests/test_containers_integration_tests/pyrefly.toml b/api/tests/test_containers_integration_tests/pyrefly.toml index b9e83741d05..e73d5bbe117 100644 --- a/api/tests/test_containers_integration_tests/pyrefly.toml +++ b/api/tests/test_containers_integration_tests/pyrefly.toml @@ -115,7 +115,6 @@ project-excludes = [ "services/test_restore_archived_workflow_run.py", "services/test_saved_message_service.py", "services/test_schedule_service.py", - "services/test_tag_service.py", "services/test_web_conversation_service.py", "services/test_webapp_auth_service.py", "services/test_webhook_service.py", diff --git a/api/tests/test_containers_integration_tests/services/test_tag_service.py b/api/tests/test_containers_integration_tests/services/test_tag_service.py index 197415ee6bd..748cca6c845 100644 --- a/api/tests/test_containers_integration_tests/services/test_tag_service.py +++ b/api/tests/test_containers_integration_tests/services/test_tag_service.py @@ -1,17 +1,19 @@ import uuid -from unittest.mock import create_autospec, patch +from collections.abc import Generator, Sequence +from dataclasses import dataclass +from unittest.mock import patch import pytest -from faker import Faker from sqlalchemy import select from sqlalchemy.orm import Session from werkzeug.exceptions import NotFound from core.rag.index_processor.constant.index_type import IndexTechniqueType -from models import Account, Tenant, TenantAccountJoin, TenantAccountRole +from models import Account, AccountStatus, Tenant, TenantAccountJoin, TenantAccountRole, TenantStatus from models.dataset import Dataset from models.enums import DataSourceType, TagType from models.model import App, Tag, TagBinding +from models.snippet import CustomizedSnippet, SnippetType from services.tag_service import ( SaveTagPayload, TagBindingCreatePayload, @@ -21,1338 +23,1044 @@ from services.tag_service import ( ) -class TestTagService: - """Integration tests for TagService using testcontainers.""" +@dataclass +class _CurrentUserStub: + id: str + current_tenant_id: str - @pytest.fixture - def mock_external_service_dependencies(self): - """Mock setup for external service dependencies.""" - with ( - patch("services.tag_service.current_user", create_autospec(Account, instance=True)) as mock_current_user, - ): - # Setup default mock returns - mock_current_user.current_tenant_id = "test-tenant-id" - mock_current_user.id = "test-user-id" - yield { - "current_user": mock_current_user, - } +@pytest.fixture +def current_user_stub() -> Generator[_CurrentUserStub, None, None]: + current_user = _CurrentUserStub(id="test-user-id", current_tenant_id="test-tenant-id") - def _create_test_account_and_tenant(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Helper method to create a test account and tenant for testing. + with patch("services.tag_service.current_user", current_user): + yield current_user - Args: - db_session_with_containers: Database session from testcontainers infrastructure - mock_external_service_dependencies: Mock dependencies - Returns: - tuple: (account, tenant) - Created account and tenant instances - """ - fake = Faker() +def _create_account_with_tenant(db_session_with_containers: Session) -> tuple[Account, Tenant]: + suffix = uuid.uuid4().hex + account = Account( + email=f"tag-service-{suffix}@example.com", + name=f"Tag Service Account {suffix}", + interface_language="en-US", + status=AccountStatus.ACTIVE, + ) - # Create account - account = Account( - email=fake.email(), - name=fake.name(), - interface_language="en-US", - status="active", - ) + tenant = Tenant( + name=f"Tag Service Tenant {suffix}", + status=TenantStatus.NORMAL, + ) + db_session_with_containers.add_all([account, tenant]) + db_session_with_containers.flush() - db_session_with_containers.add(account) - db_session_with_containers.commit() + join = TenantAccountJoin( + tenant_id=tenant.id, + account_id=account.id, + role=TenantAccountRole.OWNER, + current=True, + ) + db_session_with_containers.add(join) + db_session_with_containers.flush() - # Create tenant for the account - tenant = Tenant( - name=fake.company(), - status="normal", - ) - db_session_with_containers.add(tenant) - db_session_with_containers.commit() + return account, tenant - # Create tenant-account join - join = TenantAccountJoin( - tenant_id=tenant.id, - account_id=account.id, - role=TenantAccountRole.OWNER, - current=True, - ) - db_session_with_containers.add(join) - db_session_with_containers.commit() - # Set current tenant for account - account.current_tenant = tenant +def _set_current_user(current_user_stub: _CurrentUserStub, account: Account, tenant: Tenant) -> None: + current_user_stub.current_tenant_id = tenant.id + current_user_stub.id = account.id - # Update mock to use real tenant ID - mock_external_service_dependencies["current_user"].current_tenant_id = tenant.id - mock_external_service_dependencies["current_user"].id = account.id - return account, tenant +def _create_dataset( + db_session_with_containers: Session, + *, + tenant_id: str, + user_id: str, +) -> Dataset: + suffix = uuid.uuid4().hex + dataset = Dataset( + name=f"Tag Service Dataset {suffix}", + description="Tag service test dataset", + provider="vendor", + permission="only_me", + data_source_type=DataSourceType.UPLOAD_FILE, + indexing_technique=IndexTechniqueType.HIGH_QUALITY, + tenant_id=tenant_id, + created_by=user_id, + ) - def _create_test_dataset(self, db_session_with_containers: Session, mock_external_service_dependencies, tenant_id): - """ - Helper method to create a test dataset for testing. + db_session_with_containers.add(dataset) + db_session_with_containers.flush() - Args: - db_session_with_containers: Database session from testcontainers infrastructure - mock_external_service_dependencies: Mock dependencies - tenant_id: Tenant ID for the dataset + return dataset - Returns: - Dataset: Created dataset instance - """ - fake = Faker() - dataset = Dataset( - name=fake.company(), - description=fake.text(max_nb_chars=100), - provider="vendor", - permission="only_me", - data_source_type=DataSourceType.UPLOAD_FILE, - indexing_technique=IndexTechniqueType.HIGH_QUALITY, +def _create_app( + db_session_with_containers: Session, + *, + tenant_id: str, + user_id: str, +) -> App: + suffix = uuid.uuid4().hex + app = App( + name=f"Tag Service App {suffix}", + description="Tag service test app", + mode="chat", + icon_type="emoji", + icon="🤖", + icon_background="#FF6B6B", + enable_site=False, + enable_api=False, + tenant_id=tenant_id, + created_by=user_id, + ) + + db_session_with_containers.add(app) + db_session_with_containers.flush() + + return app + + +def _create_snippet( + db_session_with_containers: Session, + *, + tenant_id: str, + user_id: str, +) -> CustomizedSnippet: + suffix = uuid.uuid4().hex + snippet = CustomizedSnippet( + tenant_id=tenant_id, + name=f"Tag Service Snippet {suffix}", + description="Tag service test snippet", + type=SnippetType.NODE.value, + created_by=user_id, + updated_by=user_id, + ) + + db_session_with_containers.add(snippet) + db_session_with_containers.flush() + + return snippet + + +def _create_tags( + db_session_with_containers: Session, + *, + tenant_id: str, + user_id: str, + tag_type: TagType, + count: int = 3, +) -> list[Tag]: + tags = [] + + for i in range(count): + tag = Tag( + name=f"tag-{tag_type.value}-{i}-{uuid.uuid4().hex}", + type=tag_type, tenant_id=tenant_id, - created_by=mock_external_service_dependencies["current_user"].id, + created_by=user_id, ) + tags.append(tag) - db_session_with_containers.add(dataset) - db_session_with_containers.commit() + db_session_with_containers.add_all(tags) + db_session_with_containers.flush() - return dataset + return tags - def _create_test_app(self, db_session_with_containers: Session, mock_external_service_dependencies, tenant_id): - """ - Helper method to create a test app for testing. - Args: - db_session_with_containers: Database session from testcontainers infrastructure - mock_external_service_dependencies: Mock dependencies - tenant_id: Tenant ID for the app +def _create_tag_bindings( + db_session_with_containers: Session, + *, + tags: Sequence[Tag], + target_id: str, + tenant_id: str, + user_id: str, +) -> list[TagBinding]: + tag_bindings = [] - Returns: - App: Created app instance - """ - fake = Faker() - - app = App( - name=fake.company(), - description=fake.text(max_nb_chars=100), - mode="chat", - icon_type="emoji", - icon="🤖", - icon_background="#FF6B6B", - enable_site=False, - enable_api=False, + for tag in tags: + tag_binding = TagBinding( + tag_id=tag.id, + target_id=target_id, tenant_id=tenant_id, - created_by=mock_external_service_dependencies["current_user"].id, + created_by=user_id, + ) + tag_bindings.append(tag_binding) + + db_session_with_containers.add_all(tag_bindings) + db_session_with_containers.flush() + + return tag_bindings + + +def test_get_tags_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=3 + ) + + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, tags=tags[:2], target_id=dataset.id, tenant_id=tenant.id, user_id=account.id + ) + + result = TagService.get_tags(db_session_with_containers, TagType.KNOWLEDGE, tenant.id) + + assert result is not None + assert len(result) == 3 + + for tag_result in result: + assert hasattr(tag_result, "id") + assert hasattr(tag_result, "type") + assert hasattr(tag_result, "name") + assert hasattr(tag_result, "binding_count") + assert tag_result.type == TagType.KNOWLEDGE + + tag_with_bindings = next((t for t in result if t.binding_count > 0), None) + assert tag_with_bindings is not None + assert tag_with_bindings.binding_count >= 1 + + +def test_get_tags_with_keyword_filter(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=3 + ) + + tags[0].name = "python_development" + tags[1].name = "machine_learning" + tags[2].name = "web_development" + db_session_with_containers.flush() + + result = TagService.get_tags(db_session_with_containers, TagType.APP, tenant.id, keyword="development") + + assert result is not None + assert len(result) == 2 + + for tag_result in result: + assert "development" in tag_result.name.lower() + + result_no_match = TagService.get_tags(db_session_with_containers, TagType.APP, tenant.id, keyword="nonexistent") + assert result_no_match == [] + + +def test_get_tags_with_special_characters_in_keyword( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag_with_percent = Tag( + name="50% discount", + type=TagType.APP, + tenant_id=tenant.id, + created_by=account.id, + ) + tag_with_percent.id = str(uuid.uuid4()) + db_session_with_containers.add(tag_with_percent) + + tag_with_underscore = Tag( + name="test_data_tag", + type=TagType.APP, + tenant_id=tenant.id, + created_by=account.id, + ) + tag_with_underscore.id = str(uuid.uuid4()) + db_session_with_containers.add(tag_with_underscore) + + tag_with_backslash = Tag( + name="path\\to\\tag", + type=TagType.APP, + tenant_id=tenant.id, + created_by=account.id, + ) + tag_with_backslash.id = str(uuid.uuid4()) + db_session_with_containers.add(tag_with_backslash) + + tag_no_match = Tag( + name="100% different", + type=TagType.APP, + tenant_id=tenant.id, + created_by=account.id, + ) + tag_no_match.id = str(uuid.uuid4()) + db_session_with_containers.add(tag_no_match) + + db_session_with_containers.flush() + + result = TagService.get_tags(db_session_with_containers, TagType.APP, tenant.id, keyword="50%") + assert len(result) == 1 + assert result[0].name == "50% discount" + + result = TagService.get_tags(db_session_with_containers, TagType.APP, tenant.id, keyword="test_data") + assert len(result) == 1 + assert result[0].name == "test_data_tag" + + result = TagService.get_tags(db_session_with_containers, TagType.APP, tenant.id, keyword="path\\to\\tag") + assert len(result) == 1 + assert result[0].name == "path\\to\\tag" + + result = TagService.get_tags(db_session_with_containers, TagType.APP, tenant.id, keyword="50%") + assert len(result) == 1 + assert all("50%" in item.name for item in result) + + +def test_get_tags_empty_result(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + result = TagService.get_tags(db_session_with_containers, TagType.KNOWLEDGE, tenant.id) + + assert result == [] + + +def test_get_target_ids_by_tag_ids_success( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=3 + ) + + datasets = [] + for i in range(2): + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + datasets.append(dataset) + tags_to_bind = tags[:2] if i == 0 else tags[2:] + _create_tag_bindings( + db_session_with_containers, tags=tags_to_bind, target_id=dataset.id, tenant_id=tenant.id, user_id=account.id ) - db_session_with_containers.add(app) - db_session_with_containers.commit() - - return app - - def _create_test_tags( - self, db_session_with_containers: Session, mock_external_service_dependencies, tenant_id, tag_type, count=3 - ): - """ - Helper method to create test tags for testing. - - Args: - db_session_with_containers: Database session from testcontainers infrastructure - mock_external_service_dependencies: Mock dependencies - tenant_id: Tenant ID for the tags - tag_type: Type of tags to create - count: Number of tags to create - - Returns: - list: List of created tag instances - """ - fake = Faker() - tags = [] - - for i in range(count): - tag = Tag( - name=f"tag_{tag_type}_{i}_{fake.word()}", - type=tag_type, - tenant_id=tenant_id, - created_by=mock_external_service_dependencies["current_user"].id, - ) - tags.append(tag) - - for tag in tags: - db_session_with_containers.add(tag) - db_session_with_containers.commit() - - return tags - - def _create_test_tag_bindings( - self, db_session_with_containers: Session, mock_external_service_dependencies, tags, target_id, tenant_id - ): - """ - Helper method to create test tag bindings for testing. - - Args: - db_session_with_containers: Database session from testcontainers infrastructure - mock_external_service_dependencies: Mock dependencies - tags: List of tags to bind - target_id: Target ID to bind tags to - tenant_id: Tenant ID for the bindings - - Returns: - list: List of created tag binding instances - """ - tag_bindings = [] - - for tag in tags: - tag_binding = TagBinding( - tag_id=tag.id, - target_id=target_id, - tenant_id=tenant_id, - created_by=mock_external_service_dependencies["current_user"].id, - ) - tag_bindings.append(tag_binding) - - for tag_binding in tag_bindings: - db_session_with_containers.add(tag_binding) - db_session_with_containers.commit() - - return tag_bindings - - def test_get_tags_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful retrieval of tags with binding count. - - This test verifies: - - Proper tag retrieval with binding count - - Correct filtering by tag type and tenant - - Proper ordering by creation date - - Binding count calculation - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 3 - ) - - # Create dataset and bind tags - dataset = self._create_test_dataset(db_session_with_containers, mock_external_service_dependencies, tenant.id) - self._create_test_tag_bindings( - db_session_with_containers, mock_external_service_dependencies, tags[:2], dataset.id, tenant.id - ) - - # Act: Execute the method under test - result = TagService.get_tags(db_session_with_containers, "knowledge", tenant.id) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 3 - - # Verify tag data structure - for tag_result in result: - assert hasattr(tag_result, "id") - assert hasattr(tag_result, "type") - assert hasattr(tag_result, "name") - assert hasattr(tag_result, "binding_count") - assert tag_result.type == "knowledge" - - # Verify binding count - tag_with_bindings = next((t for t in result if t.binding_count > 0), None) - assert tag_with_bindings is not None - assert tag_with_bindings.binding_count >= 1 - - # Verify ordering (newest first) - note: created_at is not in SELECT but used in ORDER BY - # The ordering is handled by the database, we just verify the results are returned - assert len(result) == 3 - - def test_get_tags_with_keyword_filter( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag retrieval with keyword filtering. - - This test verifies: - - Proper keyword filtering functionality - - Case-insensitive search - - Partial match functionality - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags with specific names - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "app", 3 - ) - - # Update tag names to make them searchable - - tags[0].name = "python_development" - tags[1].name = "machine_learning" - tags[2].name = "web_development" - db_session_with_containers.commit() - - # Act: Execute the method under test with keyword filter - result = TagService.get_tags(db_session_with_containers, "app", tenant.id, keyword="development") - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 2 # Should find python_development and web_development - - # Verify filtered results contain the keyword - for tag_result in result: - assert "development" in tag_result.name.lower() - - # Verify no results for non-matching keyword - result_no_match = TagService.get_tags(db_session_with_containers, "app", tenant.id, keyword="nonexistent") - assert len(result_no_match) == 0 - - def test_get_tags_with_special_characters_in_keyword( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - r""" - Test tag retrieval with special characters in keyword to verify SQL injection prevention. - - This test verifies: - - Special characters (%, _, \) in keyword are properly escaped - - Search treats special characters as literal characters, not wildcards - - SQL injection via LIKE wildcards is prevented - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags with special characters in names - tag_with_percent = Tag( - name="50% discount", - type="app", - tenant_id=tenant.id, - created_by=account.id, - ) - tag_with_percent.id = str(uuid.uuid4()) - db_session_with_containers.add(tag_with_percent) - - tag_with_underscore = Tag( - name="test_data_tag", - type="app", - tenant_id=tenant.id, - created_by=account.id, - ) - tag_with_underscore.id = str(uuid.uuid4()) - db_session_with_containers.add(tag_with_underscore) - - tag_with_backslash = Tag( - name="path\\to\\tag", - type="app", - tenant_id=tenant.id, - created_by=account.id, - ) - tag_with_backslash.id = str(uuid.uuid4()) - db_session_with_containers.add(tag_with_backslash) - - # Create tag that should NOT match - tag_no_match = Tag( - name="100% different", - type="app", - tenant_id=tenant.id, - created_by=account.id, - ) - tag_no_match.id = str(uuid.uuid4()) - db_session_with_containers.add(tag_no_match) - - db_session_with_containers.commit() - - # Act & Assert: Test 1 - Search with % character - result = TagService.get_tags(db_session_with_containers, "app", tenant.id, keyword="50%") - assert len(result) == 1 - assert result[0].name == "50% discount" - - # Test 2 - Search with _ character - result = TagService.get_tags(db_session_with_containers, "app", tenant.id, keyword="test_data") - assert len(result) == 1 - assert result[0].name == "test_data_tag" - - # Test 3 - Search with \ character - result = TagService.get_tags(db_session_with_containers, "app", tenant.id, keyword="path\\to\\tag") - assert len(result) == 1 - assert result[0].name == "path\\to\\tag" - - # Test 4 - Search with % should NOT match 100% (verifies escaping works) - result = TagService.get_tags(db_session_with_containers, "app", tenant.id, keyword="50%") - assert len(result) == 1 - assert all("50%" in item.name for item in result) - - def test_get_tags_empty_result(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test tag retrieval when no tags exist. - - This test verifies: - - Proper handling of empty tag sets - - Correct return value for no results - """ - # Arrange: Create test data without tags - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Act: Execute the method under test - result = TagService.get_tags(db_session_with_containers, "knowledge", tenant.id) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 0 - assert isinstance(result, list) - - def test_get_target_ids_by_tag_ids_success( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test successful retrieval of target IDs by tag IDs. - - This test verifies: - - Proper target ID retrieval for valid tag IDs - - Correct filtering by tag type and tenant - - Proper handling of tag bindings - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 3 - ) - - # Create multiple datasets and bind tags - datasets = [] - for i in range(2): - dataset = self._create_test_dataset( - db_session_with_containers, mock_external_service_dependencies, tenant.id - ) - datasets.append(dataset) - # Bind first two tags to first dataset, last tag to second dataset - tags_to_bind = tags[:2] if i == 0 else tags[2:] - self._create_test_tag_bindings( - db_session_with_containers, mock_external_service_dependencies, tags_to_bind, dataset.id, tenant.id - ) - - # Act: Execute the method under test - tag_ids = [tag.id for tag in tags] - result = TagService.get_target_ids_by_tag_ids("knowledge", tenant.id, tag_ids, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 3 # Should find 3 target IDs (2 from first dataset, 1 from second) - - # Verify all dataset IDs are returned - dataset_ids = [dataset.id for dataset in datasets] - for target_id in result: - assert target_id in dataset_ids - - # Verify the first dataset appears twice (for the first two tags) - first_dataset_count = result.count(datasets[0].id) - assert first_dataset_count == 2 - - # Verify the second dataset appears once (for the last tag) - second_dataset_count = result.count(datasets[1].id) - assert second_dataset_count == 1 - - def test_get_target_ids_by_tag_ids_empty_tag_ids( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test target ID retrieval with empty tag IDs list. - - This test verifies: - - Proper handling of empty tag IDs - - Correct return value for empty input - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Act: Execute the method under test with empty tag IDs - result = TagService.get_target_ids_by_tag_ids("knowledge", tenant.id, [], db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 0 - assert isinstance(result, list) - - def test_get_target_ids_by_tag_ids_match_all( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test target ID retrieval when every requested tag must be bound to the same target. - - This test verifies: - - Targets with only one requested tag are excluded - - Targets with all requested tags are returned once - - Missing requested tags make the filter unsatisfiable - """ - # Arrange: Create test data - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 2 - ) - dataset_with_all_tags = self._create_test_dataset( - db_session_with_containers, mock_external_service_dependencies, tenant.id - ) - dataset_with_one_tag = self._create_test_dataset( - db_session_with_containers, mock_external_service_dependencies, tenant.id - ) - self._create_test_tag_bindings( - db_session_with_containers, - mock_external_service_dependencies, - tags, - dataset_with_all_tags.id, - tenant.id, - ) - self._create_test_tag_bindings( - db_session_with_containers, - mock_external_service_dependencies, - tags[:1], - dataset_with_one_tag.id, - tenant.id, - ) - - # Act: Execute the method under test - tag_ids = [tag.id for tag in tags] - result = TagService.get_target_ids_by_tag_ids( - "knowledge", tenant.id, tag_ids, db_session_with_containers, match_all=True - ) - - # Assert: Verify the expected outcomes - assert result == [dataset_with_all_tags.id] - - missing_tag_result = TagService.get_target_ids_by_tag_ids( - "knowledge", - tenant.id, - [tags[0].id, str(uuid.uuid4())], - db_session_with_containers, - match_all=True, - ) - assert missing_tag_result == [] - - def test_get_target_ids_by_tag_ids_no_matching_tags( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test target ID retrieval when no tags match the criteria. - - This test verifies: - - Proper handling of non-existent tag IDs - - Correct return value for no matches - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent tag IDs - import uuid - - non_existent_tag_ids = [str(uuid.uuid4()), str(uuid.uuid4())] - - # Act: Execute the method under test - result = TagService.get_target_ids_by_tag_ids( - "knowledge", tenant.id, non_existent_tag_ids, db_session_with_containers - ) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 0 - assert isinstance(result, list) - - def test_get_tag_by_tag_name_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful retrieval of tags by tag name. - - This test verifies: - - Proper tag retrieval by name - - Correct filtering by tag type and tenant - - Proper return value structure - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags with specific names - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "app", 2 - ) - - # Update tag names to make them searchable - - tags[0].name = "python_tag" - tags[1].name = "ml_tag" - db_session_with_containers.commit() - - # Act: Execute the method under test - result = TagService.get_tag_by_tag_name("app", tenant.id, "python_tag", db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 1 - assert result[0].name == "python_tag" - assert result[0].type == TagType.APP - assert result[0].tenant_id == tenant.id - - def test_get_tag_by_tag_name_no_matches( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag retrieval by name when no matches exist. - - This test verifies: - - Proper handling of non-existent tag names - - Correct return value for no matches - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Act: Execute the method under test with non-existent tag name - result = TagService.get_tag_by_tag_name("knowledge", tenant.id, "nonexistent_tag", db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 0 - assert isinstance(result, list) - - def test_get_tag_by_tag_name_empty_parameters( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag retrieval by name with empty parameters. - - This test verifies: - - Proper handling of empty tag type - - Proper handling of empty tag name - - Correct return value for invalid input - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Act: Execute the method under test with empty parameters - result_empty_type = TagService.get_tag_by_tag_name("", tenant.id, "test_tag", db_session_with_containers) - result_empty_name = TagService.get_tag_by_tag_name("knowledge", tenant.id, "", db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result_empty_type is not None - assert len(result_empty_type) == 0 - assert result_empty_name is not None - assert len(result_empty_name) == 0 - - def test_get_tags_by_target_id_success( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test successful retrieval of tags by target ID. - - This test verifies: - - Proper tag retrieval for a specific target - - Correct filtering by tag type and tenant - - Proper join with tag bindings - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "app", 3 - ) - - # Create app and bind tags - app = self._create_test_app(db_session_with_containers, mock_external_service_dependencies, tenant.id) - self._create_test_tag_bindings( - db_session_with_containers, mock_external_service_dependencies, tags, app.id, tenant.id - ) - - # Act: Execute the method under test - result = TagService.get_tags_by_target_id("app", tenant.id, app.id, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 3 - - # Verify all tags are returned - for tag in result: - assert tag.type == TagType.APP - assert tag.tenant_id == tenant.id - assert tag.id in [t.id for t in tags] - - def test_get_tags_by_target_id_no_bindings( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag retrieval by target ID when no tags are bound. - - This test verifies: - - Proper handling of targets with no tag bindings - - Correct return value for no results - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create app without binding any tags - app = self._create_test_app(db_session_with_containers, mock_external_service_dependencies, tenant.id) - - # Act: Execute the method under test - result = TagService.get_tags_by_target_id("app", tenant.id, app.id, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 0 - assert isinstance(result, list) - - def test_save_tags_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful tag creation. - - This test verifies: - - Proper tag creation with all required fields - - Correct database state after creation - - Proper UUID generation - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - tag_args = SaveTagPayload(name="test_tag_name", type="knowledge") - - # Act: Execute the method under test - result = TagService.save_tags(tag_args, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert result.name == "test_tag_name" - assert result.type == "knowledge" - assert result.tenant_id == tenant.id - assert result.created_by == account.id - assert result.id is not None - - # Verify database state - - db_session_with_containers.refresh(result) - assert result.id is not None - - # Verify tag was actually saved to database - saved_tag = db_session_with_containers.query(Tag).where(Tag.id == result.id).first() - assert saved_tag is not None - assert saved_tag.name == "test_tag_name" - - def test_save_tags_duplicate_name_error( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag creation with duplicate name. - - This test verifies: - - Proper error handling for duplicate tag names - - Correct exception type and message - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create first tag - tag_args = SaveTagPayload(name="duplicate_tag", type="app") + tag_ids = [tag.id for tag in tags] + result = TagService.get_target_ids_by_tag_ids(TagType.KNOWLEDGE, tenant.id, tag_ids, db_session_with_containers) + + assert result is not None + assert len(result) == 3 + + dataset_ids = [dataset.id for dataset in datasets] + for target_id in result: + assert target_id in dataset_ids + + first_dataset_count = result.count(datasets[0].id) + assert first_dataset_count == 2 + + second_dataset_count = result.count(datasets[1].id) + assert second_dataset_count == 1 + + +def test_get_target_ids_by_tag_ids_empty_tag_ids( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + result = TagService.get_target_ids_by_tag_ids(TagType.KNOWLEDGE, tenant.id, [], db_session_with_containers) + + assert result == [] + + +def test_get_target_ids_by_tag_ids_empty_snippet_tag_ids( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + result = TagService.get_target_ids_by_tag_ids(TagType.SNIPPET, tenant.id, [], db_session_with_containers) + + assert result == [] + + +def test_get_target_ids_by_tag_ids_match_all( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=2 + ) + dataset_with_all_tags = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + dataset_with_one_tag = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, + tags=tags, + target_id=dataset_with_all_tags.id, + tenant_id=tenant.id, + user_id=account.id, + ) + _create_tag_bindings( + db_session_with_containers, + tags=tags[:1], + target_id=dataset_with_one_tag.id, + tenant_id=tenant.id, + user_id=account.id, + ) + + tag_ids = [tag.id for tag in tags] + result = TagService.get_target_ids_by_tag_ids( + TagType.KNOWLEDGE, tenant.id, tag_ids, db_session_with_containers, match_all=True + ) + + assert result == [dataset_with_all_tags.id] + + missing_tag_result = TagService.get_target_ids_by_tag_ids( + TagType.KNOWLEDGE, + tenant.id, + [tags[0].id, str(uuid.uuid4())], + db_session_with_containers, + match_all=True, + ) + assert missing_tag_result == [] + + +def test_get_target_ids_by_tag_ids_no_matching_tags( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + non_existent_tag_ids = [str(uuid.uuid4()), str(uuid.uuid4())] + + result = TagService.get_target_ids_by_tag_ids( + TagType.KNOWLEDGE, tenant.id, non_existent_tag_ids, db_session_with_containers + ) + + assert result == [] + + +def test_get_tag_by_tag_name_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=2 + ) + + tags[0].name = "python_tag" + tags[1].name = "ml_tag" + db_session_with_containers.flush() + + result = TagService.get_tag_by_tag_name(TagType.APP, tenant.id, "python_tag", db_session_with_containers) + + assert result is not None + assert len(result) == 1 + assert result[0].name == "python_tag" + assert result[0].type == TagType.APP + assert result[0].tenant_id == tenant.id + + +def test_get_tag_by_tag_name_no_matches( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + result = TagService.get_tag_by_tag_name(TagType.KNOWLEDGE, tenant.id, "nonexistent_tag", db_session_with_containers) + + assert result == [] + + +def test_get_tag_by_tag_name_empty_parameters( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + result_empty_type = TagService.get_tag_by_tag_name("", tenant.id, "test_tag", db_session_with_containers) + result_empty_name = TagService.get_tag_by_tag_name(TagType.KNOWLEDGE, tenant.id, "", db_session_with_containers) + + assert result_empty_type is not None + assert len(result_empty_type) == 0 + assert result_empty_name is not None + assert len(result_empty_name) == 0 + + +def test_get_tags_by_target_id_success( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=3 + ) + + app = _create_app(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, tags=tags, target_id=app.id, tenant_id=tenant.id, user_id=account.id + ) + + result = TagService.get_tags_by_target_id(TagType.APP, tenant.id, app.id, db_session_with_containers) + + assert result is not None + assert len(result) == 3 + + for tag in result: + assert tag.type == TagType.APP + assert tag.tenant_id == tenant.id + assert tag.id in [t.id for t in tags] + + +def test_get_tags_by_target_id_no_bindings( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + app = _create_app(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + result = TagService.get_tags_by_target_id(TagType.APP, tenant.id, app.id, db_session_with_containers) + + assert result is not None + assert len(result) == 0 + assert isinstance(result, list) + + +def test_save_tags_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag_args = SaveTagPayload(name="test_tag_name", type=TagType.KNOWLEDGE) + + result = TagService.save_tags(tag_args, db_session_with_containers) + + assert result is not None + assert result.name == "test_tag_name" + assert result.type == TagType.KNOWLEDGE + assert result.tenant_id == tenant.id + assert result.created_by == account.id + assert result.id is not None + + db_session_with_containers.refresh(result) + assert result.id is not None + + saved_tag = db_session_with_containers.scalar(select(Tag).where(Tag.id == result.id)) + assert saved_tag is not None + assert saved_tag.name == "test_tag_name" + + +def test_save_tags_duplicate_name_error( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag_args = SaveTagPayload(name="duplicate_tag", type=TagType.APP) + TagService.save_tags(tag_args, db_session_with_containers) + + with pytest.raises(ValueError, match="Tag name already exists"): TagService.save_tags(tag_args, db_session_with_containers) - # Act & Assert: Verify proper error handling - with pytest.raises(ValueError) as exc_info: - TagService.save_tags(tag_args, db_session_with_containers) - assert "Tag name already exists" in str(exc_info.value) - def test_update_tags_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful tag update. +def test_update_tags_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) - This test verifies: - - Proper tag update with new name - - Correct database state after update - - Proper error handling for non-existent tags - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies + tag_args = SaveTagPayload(name="original_name", type=TagType.KNOWLEDGE) + tag = TagService.save_tags(tag_args, db_session_with_containers) + + update_args = UpdateTagPayload(name="updated_name") + + result = TagService.update_tags(update_args, tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) + + assert result is not None + assert result.name == "updated_name" + assert result.type == TagType.KNOWLEDGE + assert result.id == tag.id + + db_session_with_containers.refresh(result) + assert result.name == "updated_name" + + updated_tag = db_session_with_containers.scalar(select(Tag).where(Tag.id == tag.id)) + assert updated_tag is not None + assert updated_tag.name == "updated_name" + + +def test_update_tags_not_found_error(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + non_existent_tag_id = str(uuid.uuid4()) + + update_args = UpdateTagPayload(name="updated_name") + + with pytest.raises(NotFound, match="Tag not found"): + TagService.update_tags(update_args, non_existent_tag_id, db_session_with_containers) + + +def test_update_tags_duplicate_name_error( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag1_args = SaveTagPayload(name="first_tag", type=TagType.APP) + TagService.save_tags(tag1_args, db_session_with_containers) + + tag2_args = SaveTagPayload(name="second_tag", type=TagType.APP) + tag2 = TagService.save_tags(tag2_args, db_session_with_containers) + + update_args = UpdateTagPayload(name="first_tag") + + with pytest.raises(ValueError, match="Tag name already exists"): + TagService.update_tags(update_args, tag2.id, db_session_with_containers) + + +def test_update_tags_requires_current_tenant_and_type( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + app_tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + other_account, other_tenant = _create_account_with_tenant(db_session_with_containers) + other_tenant_tag = _create_tags( + db_session_with_containers, + tenant_id=other_tenant.id, + user_id=other_account.id, + tag_type=TagType.KNOWLEDGE, + count=1, + )[0] + app_tag_name = app_tag.name + other_tenant_tag_name = other_tenant_tag.name + update_args = UpdateTagPayload(name="should_not_update") + + with pytest.raises(NotFound, match="Tag not found"): + TagService.update_tags(update_args, app_tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) + + with pytest.raises(NotFound, match="Tag not found"): + TagService.update_tags(update_args, other_tenant_tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) + + db_session_with_containers.refresh(app_tag) + db_session_with_containers.refresh(other_tenant_tag) + assert app_tag.name == app_tag_name + assert other_tenant_tag.name == other_tenant_tag_name + + +def test_get_tag_binding_count_success( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=2 + ) + + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, tags=[tags[0]], target_id=dataset.id, tenant_id=tenant.id, user_id=account.id + ) + + result_tag_with_bindings = TagService.get_tag_binding_count( + tags[0].id, db_session_with_containers, tag_type=TagType.KNOWLEDGE + ) + result_tag_without_bindings = TagService.get_tag_binding_count( + tags[1].id, db_session_with_containers, tag_type=TagType.KNOWLEDGE + ) + + assert result_tag_with_bindings == 1 + assert result_tag_without_bindings == 0 + + +def test_get_tag_binding_count_scopes_to_current_tenant_and_type( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=1 + )[0] + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, tags=[tag], target_id=dataset.id, tenant_id=tenant.id, user_id=account.id + ) + other_account, other_tenant = _create_account_with_tenant(db_session_with_containers) + other_tenant_tag = _create_tags( + db_session_with_containers, + tenant_id=other_tenant.id, + user_id=other_account.id, + tag_type=TagType.KNOWLEDGE, + count=1, + )[0] + other_tenant_dataset = _create_dataset( + db_session_with_containers, tenant_id=other_tenant.id, user_id=other_account.id + ) + _create_tag_bindings( + db_session_with_containers, + tags=[other_tenant_tag], + target_id=other_tenant_dataset.id, + tenant_id=other_tenant.id, + user_id=other_account.id, + ) + + assert TagService.get_tag_binding_count(tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) == 1 + assert TagService.get_tag_binding_count(tag.id, db_session_with_containers, tag_type=TagType.APP) == 0 + assert ( + TagService.get_tag_binding_count(other_tenant_tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) + == 0 + ) + + +def test_get_tag_binding_count_non_existent_tag( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + non_existent_tag_id = str(uuid.uuid4()) + + result = TagService.get_tag_binding_count(non_existent_tag_id, db_session_with_containers) + + assert result == 0 + + +def test_delete_tag_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + + app = _create_app(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, tags=[tag], target_id=app.id, tenant_id=tenant.id, user_id=account.id + ) + + tag_before = db_session_with_containers.scalar(select(Tag).where(Tag.id == tag.id)) + assert tag_before is not None + + binding_before = db_session_with_containers.scalar(select(TagBinding).where(TagBinding.tag_id == tag.id)) + assert binding_before is not None + + TagService.delete_tag(tag.id, db_session_with_containers, tag_type=TagType.APP) + + tag_after = db_session_with_containers.scalar(select(Tag).where(Tag.id == tag.id)) + assert tag_after is None + + binding_after = db_session_with_containers.scalar(select(TagBinding).where(TagBinding.tag_id == tag.id)) + assert binding_after is None + + +def test_delete_tag_not_found_error(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + non_existent_tag_id = str(uuid.uuid4()) + + with pytest.raises(NotFound, match="Tag not found"): + TagService.delete_tag(non_existent_tag_id, db_session_with_containers) + + +def test_delete_tag_requires_current_tenant_and_type( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + app_tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + other_account, other_tenant = _create_account_with_tenant(db_session_with_containers) + other_tenant_tag = _create_tags( + db_session_with_containers, + tenant_id=other_tenant.id, + user_id=other_account.id, + tag_type=TagType.KNOWLEDGE, + count=1, + )[0] + + with pytest.raises(NotFound, match="Tag not found"): + TagService.delete_tag(app_tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) + + with pytest.raises(NotFound, match="Tag not found"): + TagService.delete_tag(other_tenant_tag.id, db_session_with_containers, tag_type=TagType.KNOWLEDGE) + + assert db_session_with_containers.scalar(select(Tag).where(Tag.id == app_tag.id)) is not None + assert db_session_with_containers.scalar(select(Tag).where(Tag.id == other_tenant_tag.id)) is not None + + +def test_save_tag_binding_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=2 + ) + + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + binding_payload = TagBindingCreatePayload( + type=TagType.KNOWLEDGE, target_id=dataset.id, tag_ids=[tag.id for tag in tags] + ) + TagService.save_tag_binding(binding_payload, db_session_with_containers) + + for tag in tags: + binding = db_session_with_containers.scalar( + select(TagBinding).where(TagBinding.tag_id == tag.id, TagBinding.target_id == dataset.id) ) + assert binding is not None + assert binding.tenant_id == tenant.id + assert binding.created_by == account.id + + +def test_save_tag_binding_only_creates_bindings_for_valid_snippet_tags( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + snippet = _create_snippet(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + valid_tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.SNIPPET, count=1 + )[0] + app_tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + other_account, other_tenant = _create_account_with_tenant(db_session_with_containers) + other_tenant_tag = _create_tags( + db_session_with_containers, + tenant_id=other_tenant.id, + user_id=other_account.id, + tag_type=TagType.SNIPPET, + count=1, + )[0] + + binding_payload = TagBindingCreatePayload( + type=TagType.SNIPPET, + target_id=snippet.id, + tag_ids=[valid_tag.id, app_tag.id, other_tenant_tag.id], + ) + TagService.save_tag_binding(binding_payload, db_session_with_containers) + + bindings = db_session_with_containers.scalars(select(TagBinding).where(TagBinding.target_id == snippet.id)).all() + assert len(bindings) == 1 + assert bindings[0].tag_id == valid_tag.id + assert bindings[0].tenant_id == tenant.id + assert bindings[0].created_by == account.id + + +def test_save_tag_binding_duplicate_handling( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + + app = _create_app(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + binding_payload = TagBindingCreatePayload(type=TagType.APP, target_id=app.id, tag_ids=[tag.id]) + TagService.save_tag_binding(binding_payload, db_session_with_containers) + + TagService.save_tag_binding(binding_payload, db_session_with_containers) + + bindings = db_session_with_containers.scalars( + select(TagBinding).where(TagBinding.tag_id == tag.id, TagBinding.target_id == app.id) + ).all() + assert len(bindings) == 1 + + +def test_delete_tag_binding_success(db_session_with_containers: Session, current_user_stub: _CurrentUserStub) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tags = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.KNOWLEDGE, count=2 + ) + + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + _create_tag_bindings( + db_session_with_containers, tags=tags, target_id=dataset.id, tenant_id=tenant.id, user_id=account.id + ) + + bindings_before = db_session_with_containers.scalars( + select(TagBinding).where(TagBinding.tag_id.in_([tag.id for tag in tags]), TagBinding.target_id == dataset.id) + ).all() + assert len(bindings_before) == 2 + + delete_payload = TagBindingDeletePayload( + type=TagType.KNOWLEDGE, target_id=dataset.id, tag_ids=[tag.id for tag in tags] + ) + TagService.delete_tag_binding(delete_payload, db_session_with_containers) + + bindings_after = db_session_with_containers.scalars( + select(TagBinding).where(TagBinding.tag_id.in_([tag.id for tag in tags]), TagBinding.target_id == dataset.id) + ).all() + assert bindings_after == [] + + +def test_delete_tag_binding_preserves_wrong_type_and_other_tenant_snippet_bindings( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + snippet = _create_snippet(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + valid_tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.SNIPPET, count=1 + )[0] + app_tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + other_account, other_tenant = _create_account_with_tenant(db_session_with_containers) + other_tenant_tag = _create_tags( + db_session_with_containers, + tenant_id=other_tenant.id, + user_id=other_account.id, + tag_type=TagType.SNIPPET, + count=1, + )[0] + _create_tag_bindings( + db_session_with_containers, tags=[valid_tag], target_id=snippet.id, tenant_id=tenant.id, user_id=account.id + ) + _create_tag_bindings( + db_session_with_containers, tags=[app_tag], target_id=snippet.id, tenant_id=tenant.id, user_id=account.id + ) + _create_tag_bindings( + db_session_with_containers, + tags=[other_tenant_tag], + target_id=snippet.id, + tenant_id=other_tenant.id, + user_id=other_account.id, + ) + + delete_payload = TagBindingDeletePayload( + type=TagType.SNIPPET, + target_id=snippet.id, + tag_ids=[valid_tag.id, app_tag.id, other_tenant_tag.id], + ) + TagService.delete_tag_binding(delete_payload, db_session_with_containers) - # Create a tag to update - tag_args = SaveTagPayload(name="original_name", type="knowledge") - tag = TagService.save_tags(tag_args, db_session_with_containers) - - # Update args - update_args = UpdateTagPayload(name="updated_name") - - # Act: Execute the method under test - result = TagService.update_tags(update_args, tag.id, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result is not None - assert result.name == "updated_name" - assert result.type == "knowledge" - assert result.id == tag.id - - # Verify database state - - db_session_with_containers.refresh(result) - assert result.name == "updated_name" - - # Verify tag was actually updated in database - updated_tag = db_session_with_containers.query(Tag).where(Tag.id == tag.id).first() - assert updated_tag is not None - assert updated_tag.name == "updated_name" - - def test_update_tags_not_found_error(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test tag update for non-existent tag. - - This test verifies: - - Proper error handling for non-existent tags - - Correct exception type - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent tag ID - import uuid - - non_existent_tag_id = str(uuid.uuid4()) - - update_args = UpdateTagPayload(name="updated_name") - - # Act & Assert: Verify proper error handling - with pytest.raises(NotFound) as exc_info: - TagService.update_tags(update_args, non_existent_tag_id, db_session_with_containers) - assert "Tag not found" in str(exc_info.value) - - def test_update_tags_duplicate_name_error( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag update with duplicate name. - - This test verifies: - - Proper error handling for duplicate tag names during update - - Correct exception type and message - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create two tags - tag1_args = SaveTagPayload(name="first_tag", type="app") - tag1 = TagService.save_tags(tag1_args, db_session_with_containers) - - tag2_args = SaveTagPayload(name="second_tag", type="app") - tag2 = TagService.save_tags(tag2_args, db_session_with_containers) - - # Try to update second tag with first tag's name - update_args = UpdateTagPayload(name="first_tag") - - # Act & Assert: Verify proper error handling - with pytest.raises(ValueError) as exc_info: - TagService.update_tags(update_args, tag2.id, db_session_with_containers) - assert "Tag name already exists" in str(exc_info.value) - - def test_get_tag_binding_count_success( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test successful retrieval of tag binding count. - - This test verifies: - - Proper binding count calculation - - Correct handling of tags with no bindings - - Proper database query execution - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 2 - ) - - # Create dataset and bind first tag - dataset = self._create_test_dataset(db_session_with_containers, mock_external_service_dependencies, tenant.id) - self._create_test_tag_bindings( - db_session_with_containers, mock_external_service_dependencies, [tags[0]], dataset.id, tenant.id - ) - - # Act: Execute the method under test - result_tag_with_bindings = TagService.get_tag_binding_count(tags[0].id, db_session_with_containers) - result_tag_without_bindings = TagService.get_tag_binding_count(tags[1].id, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result_tag_with_bindings == 1 - assert result_tag_without_bindings == 0 - - def test_get_tag_binding_count_non_existent_tag( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test binding count retrieval for non-existent tag. - - This test verifies: - - Proper handling of non-existent tag IDs - - Correct return value for non-existent tags - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent tag ID - import uuid - - non_existent_tag_id = str(uuid.uuid4()) - - # Act: Execute the method under test - result = TagService.get_tag_binding_count(non_existent_tag_id, db_session_with_containers) - - # Assert: Verify the expected outcomes - assert result == 0 - - def test_delete_tag_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful tag deletion. - - This test verifies: - - Proper tag deletion from database - - Proper cleanup of associated tag bindings - - Correct database state after deletion - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tag with bindings - tag = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "app", 1 - )[0] - - # Create app and bind tag - app = self._create_test_app(db_session_with_containers, mock_external_service_dependencies, tenant.id) - self._create_test_tag_bindings( - db_session_with_containers, mock_external_service_dependencies, [tag], app.id, tenant.id - ) - - # Verify tag and binding exist before deletion - - tag_before = db_session_with_containers.query(Tag).where(Tag.id == tag.id).first() - assert tag_before is not None - - binding_before = db_session_with_containers.query(TagBinding).where(TagBinding.tag_id == tag.id).first() - assert binding_before is not None - - # Act: Execute the method under test - TagService.delete_tag(tag.id, db_session_with_containers) - - # Assert: Verify the expected outcomes - # Verify tag was deleted - tag_after = db_session_with_containers.query(Tag).where(Tag.id == tag.id).first() - assert tag_after is None - - # Verify tag binding was deleted - binding_after = db_session_with_containers.query(TagBinding).where(TagBinding.tag_id == tag.id).first() - assert binding_after is None - - def test_delete_tag_not_found_error(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test tag deletion for non-existent tag. - - This test verifies: - - Proper error handling for non-existent tags - - Correct exception type - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent tag ID - import uuid - - non_existent_tag_id = str(uuid.uuid4()) - - # Act & Assert: Verify proper error handling - with pytest.raises(NotFound) as exc_info: - TagService.delete_tag(non_existent_tag_id, db_session_with_containers) - assert "Tag not found" in str(exc_info.value) - - def test_save_tag_binding_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful tag binding creation. - - This test verifies: - - Proper tag binding creation - - Correct handling of duplicate bindings - - Proper database state after creation - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 2 - ) - - # Create dataset - dataset = self._create_test_dataset(db_session_with_containers, mock_external_service_dependencies, tenant.id) - - # Act: Execute the method under test - binding_payload = TagBindingCreatePayload( - type="knowledge", target_id=dataset.id, tag_ids=[tag.id for tag in tags] - ) - TagService.save_tag_binding(binding_payload, db_session_with_containers) - - # Assert: Verify the expected outcomes - - # Verify tag bindings were created - for tag in tags: - binding = ( - db_session_with_containers.query(TagBinding) - .where(TagBinding.tag_id == tag.id, TagBinding.target_id == dataset.id) - .first() - ) - assert binding is not None - assert binding.tenant_id == tenant.id - assert binding.created_by == account.id - - def test_save_tag_binding_duplicate_handling( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag binding creation with duplicate bindings. - - This test verifies: - - Proper handling of duplicate tag bindings - - No errors when trying to create existing bindings - - Correct database state after operation - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tag - tag = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "app", 1 - )[0] - - # Create app - app = self._create_test_app(db_session_with_containers, mock_external_service_dependencies, tenant.id) - - # Create first binding - binding_payload = TagBindingCreatePayload(type="app", target_id=app.id, tag_ids=[tag.id]) - TagService.save_tag_binding(binding_payload, db_session_with_containers) - - # Act: Try to create duplicate binding - TagService.save_tag_binding(binding_payload, db_session_with_containers) - - # Assert: Verify the expected outcomes - - # Verify only one binding exists - bindings = db_session_with_containers.scalars( - select(TagBinding).where(TagBinding.tag_id == tag.id, TagBinding.target_id == app.id) - ).all() - assert len(bindings) == 1 - - def test_save_tag_binding_invalid_target_type( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag binding creation with invalid target type. - - This test verifies: - - Proper error handling for invalid target types - - Correct exception type - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tag - tag = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 1 - )[0] - - # Create non-existent target ID - import uuid - - non_existent_target_id = str(uuid.uuid4()) - - # Act & Assert: Verify proper error handling - from pydantic import ValidationError - - with pytest.raises(ValidationError): - TagBindingCreatePayload(type="invalid_type", target_id=non_existent_target_id, tag_ids=[tag.id]) - - def test_delete_tag_binding_success(self, db_session_with_containers: Session, mock_external_service_dependencies): - """ - Test successful tag binding deletion. - - This test verifies: - - Proper tag binding deletion from database - - Correct database state after deletion - - Proper error handling for non-existent bindings - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tags - tags = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "knowledge", 2 - ) - - # Create dataset and bind tags - dataset = self._create_test_dataset(db_session_with_containers, mock_external_service_dependencies, tenant.id) - self._create_test_tag_bindings( - db_session_with_containers, mock_external_service_dependencies, tags, dataset.id, tenant.id - ) - - # Verify bindings exist before deletion - bindings_before = ( - db_session_with_containers.query(TagBinding) - .where(TagBinding.tag_id.in_([tag.id for tag in tags]), TagBinding.target_id == dataset.id) - .all() - ) - assert len(bindings_before) == 2 - - # Act: Execute the method under test - delete_payload = TagBindingDeletePayload( - type="knowledge", target_id=dataset.id, tag_ids=[tag.id for tag in tags] - ) - TagService.delete_tag_binding(delete_payload, db_session_with_containers) - - # Assert: Verify the expected outcomes - # Verify tag bindings were deleted - bindings_after = ( - db_session_with_containers.query(TagBinding) - .where(TagBinding.tag_id.in_([tag.id for tag in tags]), TagBinding.target_id == dataset.id) - .all() - ) - assert len(bindings_after) == 0 - - def test_delete_tag_binding_non_existent_binding( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test tag binding deletion for non-existent binding. - - This test verifies: - - Proper handling of non-existent tag bindings - - No errors when trying to delete non-existent bindings - - Correct database state after operation - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create tag and dataset without binding - tag = self._create_test_tags( - db_session_with_containers, mock_external_service_dependencies, tenant.id, "app", 1 - )[0] - app = self._create_test_app(db_session_with_containers, mock_external_service_dependencies, tenant.id) - - # Act: Try to delete non-existent binding - delete_payload = TagBindingDeletePayload(type="app", target_id=app.id, tag_ids=[tag.id]) - TagService.delete_tag_binding(delete_payload, db_session_with_containers) - - # Assert: Verify the expected outcomes - # No error should be raised, and database state should remain unchanged - - bindings = db_session_with_containers.scalars( - select(TagBinding).where(TagBinding.tag_id == tag.id, TagBinding.target_id == app.id) - ).all() - assert len(bindings) == 0 - - def test_check_target_exists_knowledge_success( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test successful target existence check for knowledge type. - - This test verifies: - - Proper validation of knowledge dataset existence - - Correct error handling for non-existent datasets - - Proper tenant filtering - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create dataset - dataset = self._create_test_dataset(db_session_with_containers, mock_external_service_dependencies, tenant.id) - - # Act: Execute the method under test - TagService.check_target_exists("knowledge", dataset.id, db_session_with_containers) - - # Assert: Verify the expected outcomes - # No exception should be raised for existing dataset - - def test_check_target_exists_knowledge_not_found( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test target existence check for non-existent knowledge dataset. - - This test verifies: - - Proper error handling for non-existent knowledge datasets - - Correct exception type and message - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent dataset ID - import uuid - - non_existent_dataset_id = str(uuid.uuid4()) - - # Act & Assert: Verify proper error handling - with pytest.raises(NotFound) as exc_info: - TagService.check_target_exists("knowledge", non_existent_dataset_id, db_session_with_containers) - assert "Dataset not found" in str(exc_info.value) - - def test_check_target_exists_app_success( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test successful target existence check for app type. - - This test verifies: - - Proper validation of app existence - - Correct error handling for non-existent apps - - Proper tenant filtering - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create app - app = self._create_test_app(db_session_with_containers, mock_external_service_dependencies, tenant.id) - - # Act: Execute the method under test - TagService.check_target_exists("app", app.id, db_session_with_containers) - - # Assert: Verify the expected outcomes - # No exception should be raised for existing app - - def test_check_target_exists_app_not_found( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test target existence check for non-existent app. - - This test verifies: - - Proper error handling for non-existent apps - - Correct exception type and message - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent app ID - import uuid - - non_existent_app_id = str(uuid.uuid4()) - - # Act & Assert: Verify proper error handling - with pytest.raises(NotFound) as exc_info: - TagService.check_target_exists("app", non_existent_app_id, db_session_with_containers) - assert "App not found" in str(exc_info.value) - - def test_check_target_exists_invalid_type( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test target existence check for invalid type. - - This test verifies: - - Proper error handling for invalid target types - - Correct exception type and message - """ - # Arrange: Create test data - fake = Faker() - account, tenant = self._create_test_account_and_tenant( - db_session_with_containers, mock_external_service_dependencies - ) - - # Create non-existent target ID - import uuid - - non_existent_target_id = str(uuid.uuid4()) - - # Act & Assert: Verify proper error handling - with pytest.raises(NotFound) as exc_info: - TagService.check_target_exists("invalid_type", non_existent_target_id, db_session_with_containers) - assert "Invalid binding type" in str(exc_info.value) + remaining_tag_ids = set( + db_session_with_containers.scalars(select(TagBinding.tag_id).where(TagBinding.target_id == snippet.id)).all() + ) + assert remaining_tag_ids == {app_tag.id, other_tenant_tag.id} + + +def test_delete_tag_binding_succeeds_when_no_snippet_rows_match( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + snippet = _create_snippet(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.SNIPPET, count=1 + )[0] + + delete_payload = TagBindingDeletePayload(type=TagType.SNIPPET, target_id=snippet.id, tag_ids=[tag.id]) + TagService.delete_tag_binding(delete_payload, db_session_with_containers) + + bindings = db_session_with_containers.scalars(select(TagBinding).where(TagBinding.target_id == snippet.id)).all() + assert bindings == [] + + +def test_delete_tag_binding_non_existent_binding( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + tag = _create_tags( + db_session_with_containers, tenant_id=tenant.id, user_id=account.id, tag_type=TagType.APP, count=1 + )[0] + app = _create_app(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + delete_payload = TagBindingDeletePayload(type=TagType.APP, target_id=app.id, tag_ids=[tag.id]) + TagService.delete_tag_binding(delete_payload, db_session_with_containers) + + bindings = db_session_with_containers.scalars( + select(TagBinding).where(TagBinding.tag_id == tag.id, TagBinding.target_id == app.id) + ).all() + assert bindings == [] + + +def test_check_target_exists_knowledge_success( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + dataset = _create_dataset(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + TagService.check_target_exists(TagType.KNOWLEDGE, dataset.id, db_session_with_containers) + + +def test_check_target_exists_knowledge_not_found( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + non_existent_dataset_id = str(uuid.uuid4()) + + with pytest.raises(NotFound, match="Dataset not found"): + TagService.check_target_exists(TagType.KNOWLEDGE, non_existent_dataset_id, db_session_with_containers) + + +def test_check_target_exists_app_success( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + app = _create_app(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + TagService.check_target_exists(TagType.APP, app.id, db_session_with_containers) + + +def test_check_target_exists_app_not_found( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + non_existent_app_id = str(uuid.uuid4()) + + with pytest.raises(NotFound, match="App not found"): + TagService.check_target_exists(TagType.APP, non_existent_app_id, db_session_with_containers) + + +def test_check_target_exists_snippet_success( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + snippet = _create_snippet(db_session_with_containers, tenant_id=tenant.id, user_id=account.id) + + TagService.check_target_exists(TagType.SNIPPET, snippet.id, db_session_with_containers) + + +def test_check_target_exists_snippet_not_found_for_missing_id( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + + with pytest.raises(NotFound, match="Snippet not found"): + TagService.check_target_exists(TagType.SNIPPET, str(uuid.uuid4()), db_session_with_containers) + + +def test_check_target_exists_snippet_not_found_for_other_tenant( + db_session_with_containers: Session, current_user_stub: _CurrentUserStub +) -> None: + account, tenant = _create_account_with_tenant(db_session_with_containers) + _set_current_user(current_user_stub, account, tenant) + other_account, other_tenant = _create_account_with_tenant(db_session_with_containers) + other_tenant_snippet = _create_snippet( + db_session_with_containers, tenant_id=other_tenant.id, user_id=other_account.id + ) + + with pytest.raises(NotFound, match="Snippet not found"): + TagService.check_target_exists(TagType.SNIPPET, other_tenant_snippet.id, db_session_with_containers) + + +def test_check_target_exists_invalid_type(db_session_with_containers: Session) -> None: + with pytest.raises(NotFound, match="Invalid binding type"): + TagService.check_target_exists("invalid_type", "target-id", db_session_with_containers) diff --git a/api/tests/unit_tests/services/test_tag_service.py b/api/tests/unit_tests/services/test_tag_service.py deleted file mode 100644 index d304ab23cea..00000000000 --- a/api/tests/unit_tests/services/test_tag_service.py +++ /dev/null @@ -1,172 +0,0 @@ -from types import SimpleNamespace - -import pytest -from pytest_mock import MockerFixture -from werkzeug.exceptions import NotFound - -from models.enums import TagType -from services.tag_service import TagBindingCreatePayload, TagBindingDeletePayload, TagService, UpdateTagPayload - - -@pytest.fixture -def current_user(mocker: MockerFixture): - user = SimpleNamespace(id="user-1", current_tenant_id="tenant-1") - mocker.patch("services.tag_service.current_user", user) - return user - - -@pytest.fixture -def db_session(mocker: MockerFixture): - mock_db = mocker.Mock() - return mock_db.session - - -def test_save_tag_binding_only_creates_bindings_for_valid_snippet_tags(mocker: MockerFixture, current_user, db_session): - mocker.patch("services.tag_service.TagService.check_target_exists") - db_session.scalars.return_value.all.return_value = ["tag-1"] - db_session.scalar.return_value = None - - TagService.save_tag_binding( - TagBindingCreatePayload( - tag_ids=["tag-1", "tag-from-other-tenant"], - target_id="snippet-1", - type=TagType.SNIPPET, - ), - db_session, - ) - - db_session.add.assert_called_once() - tag_binding = db_session.add.call_args.args[0] - assert tag_binding.tag_id == "tag-1" - assert tag_binding.target_id == "snippet-1" - assert tag_binding.tenant_id == current_user.current_tenant_id - assert tag_binding.created_by == current_user.id - db_session.commit.assert_called_once() - - -def test_delete_tag_binding_limits_deletion_to_valid_snippet_tags(mocker: MockerFixture, current_user, db_session): - mocker.patch("services.tag_service.TagService.check_target_exists") - db_session.execute.return_value = SimpleNamespace(rowcount=1) - - TagService.delete_tag_binding( - TagBindingDeletePayload( - tag_ids=["tag-1", "tag-from-other-tenant"], - target_id="snippet-1", - type=TagType.SNIPPET, - ), - db_session, - ) - - db_session.execute.assert_called_once() - db_session.commit.assert_called_once() - - -def test_delete_tag_binding_does_not_commit_when_no_rows_deleted(mocker: MockerFixture, current_user, db_session): - mocker.patch("services.tag_service.TagService.check_target_exists") - db_session.execute.return_value = SimpleNamespace(rowcount=0) - - TagService.delete_tag_binding( - TagBindingDeletePayload( - tag_ids=["tag-1"], - target_id="snippet-1", - type=TagType.SNIPPET, - ), - db_session, - ) - - db_session.execute.assert_called_once() - db_session.commit.assert_not_called() - - -def test_update_tags_scopes_lookup_to_current_tenant_and_type(current_user, db_session): - tag = SimpleNamespace(id="tag-1", name="old", type=TagType.KNOWLEDGE) - db_session.scalar.side_effect = [tag, None] - - result = TagService.update_tags(UpdateTagPayload(name="new"), "tag-1", db_session, tag_type=TagType.KNOWLEDGE) - - stmt = db_session.scalar.call_args_list[0].args[0] - compiled = stmt.compile() - statement = str(compiled) - assert "tags.id" in statement - assert "tags.tenant_id" in statement - assert "tags.type" in statement - assert "tag-1" in compiled.params.values() - assert current_user.current_tenant_id in compiled.params.values() - assert TagType.KNOWLEDGE in compiled.params.values() - assert result is tag - assert tag.name == "new" - db_session.commit.assert_called_once() - - -def test_get_tag_binding_count_scopes_lookup_to_current_tenant_and_type(current_user, db_session): - db_session.scalar.return_value = 3 - - result = TagService.get_tag_binding_count("tag-1", db_session, tag_type=TagType.KNOWLEDGE) - - stmt = db_session.scalar.call_args.args[0] - compiled = stmt.compile() - statement = str(compiled) - assert "tag_bindings.tag_id" in statement - assert "tags.tenant_id" in statement - assert "tags.type" in statement - assert "tag-1" in compiled.params.values() - assert current_user.current_tenant_id in compiled.params.values() - assert TagType.KNOWLEDGE in compiled.params.values() - assert result == 3 - - -def test_delete_tag_scopes_lookup_and_bindings_to_current_tenant(current_user, db_session): - tag = SimpleNamespace(id="tag-1", name="old", type=TagType.KNOWLEDGE) - binding = SimpleNamespace(id="binding-1") - db_session.scalar.return_value = tag - db_session.scalars.return_value.all.return_value = [binding] - - TagService.delete_tag("tag-1", db_session, tag_type=TagType.KNOWLEDGE) - - tag_stmt = db_session.scalar.call_args.args[0] - tag_compiled = tag_stmt.compile() - assert "tags.id" in str(tag_compiled) - assert "tags.tenant_id" in str(tag_compiled) - assert "tags.type" in str(tag_compiled) - assert "tag-1" in tag_compiled.params.values() - assert current_user.current_tenant_id in tag_compiled.params.values() - assert TagType.KNOWLEDGE in tag_compiled.params.values() - - binding_stmt = db_session.scalars.call_args.args[0] - binding_compiled = binding_stmt.compile() - assert "tag_bindings.tag_id" in str(binding_compiled) - assert "tag_bindings.tenant_id" in str(binding_compiled) - assert "tag-1" in binding_compiled.params.values() - assert current_user.current_tenant_id in binding_compiled.params.values() - db_session.delete.assert_any_call(tag) - db_session.delete.assert_any_call(binding) - db_session.commit.assert_called_once() - - -def test_get_target_ids_by_tag_ids_returns_empty_without_query_for_empty_input(db_session): - result = TagService.get_target_ids_by_tag_ids(TagType.SNIPPET, "tenant-1", [], db_session) - - assert result == [] - db_session.scalars.assert_not_called() - - -def test_check_target_exists_accepts_existing_snippet(current_user, db_session): - db_session.scalar.return_value = SimpleNamespace(id="snippet-1") - - TagService.check_target_exists("snippet", "snippet-1", db_session) - - db_session.scalar.assert_called_once() - - -def test_check_target_exists_raises_when_snippet_missing(current_user, db_session): - db_session.scalar.return_value = None - - with pytest.raises(NotFound, match="Snippet not found"): - TagService.check_target_exists("snippet", "missing-snippet", db_session) - - -def test_check_target_exists_raises_for_invalid_binding_type(current_user, db_session): - with pytest.raises(NotFound, match="Invalid binding type"): - TagService.check_target_exists("unknown", "target-1", db_session) - - db_session.scalar.assert_not_called()