From 6dd0f99b90c2ec8363514d090e1c3208a14e8bb3 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 28 Dec 2025 23:35:09 +0400 Subject: [PATCH] rm: outdated tests --- .../test/apps/webui/routers/test_chats.py | 236 ------------------ .../test/util/abstract_integration_test.py | 161 ------------ 2 files changed, 397 deletions(-) delete mode 100644 backend/open_webui/test/apps/webui/routers/test_chats.py delete mode 100644 backend/open_webui/test/util/abstract_integration_test.py diff --git a/backend/open_webui/test/apps/webui/routers/test_chats.py b/backend/open_webui/test/apps/webui/routers/test_chats.py deleted file mode 100644 index a36a01fb14..0000000000 --- a/backend/open_webui/test/apps/webui/routers/test_chats.py +++ /dev/null @@ -1,236 +0,0 @@ -import uuid - -from test.util.abstract_integration_test import AbstractPostgresTest -from test.util.mock_user import mock_webui_user - - -class TestChats(AbstractPostgresTest): - BASE_PATH = "/api/v1/chats" - - def setup_class(cls): - super().setup_class() - - def setup_method(self): - super().setup_method() - from open_webui.models.chats import ChatForm, Chats - - self.chats = Chats - self.chats.insert_new_chat( - "2", - ChatForm( - **{ - "chat": { - "name": "chat1", - "description": "chat1 description", - "tags": ["tag1", "tag2"], - "history": {"currentId": "1", "messages": []}, - } - } - ), - ) - - def test_get_session_user_chat_list(self): - with mock_webui_user(id="2"): - response = self.fast_api_client.get(self.create_url("/")) - assert response.status_code == 200 - first_chat = response.json()[0] - assert first_chat["id"] is not None - assert first_chat["title"] == "New Chat" - assert first_chat["created_at"] is not None - assert first_chat["updated_at"] is not None - - def test_delete_all_user_chats(self): - with mock_webui_user(id="2"): - response = self.fast_api_client.delete(self.create_url("/")) - assert response.status_code == 200 - assert len(self.chats.get_chats()) == 0 - - def test_get_user_chat_list_by_user_id(self): - with mock_webui_user(id="3"): - response = self.fast_api_client.get(self.create_url("/list/user/2")) - assert response.status_code == 200 - first_chat = response.json()[0] - assert first_chat["id"] is not None - assert first_chat["title"] == "New Chat" - assert first_chat["created_at"] is not None - assert first_chat["updated_at"] is not None - - def test_create_new_chat(self): - with mock_webui_user(id="2"): - response = self.fast_api_client.post( - self.create_url("/new"), - json={ - "chat": { - "name": "chat2", - "description": "chat2 description", - "tags": ["tag1", "tag2"], - } - }, - ) - assert response.status_code == 200 - data = response.json() - assert data["archived"] is False - assert data["chat"] == { - "name": "chat2", - "description": "chat2 description", - "tags": ["tag1", "tag2"], - } - assert data["user_id"] == "2" - assert data["id"] is not None - assert data["share_id"] is None - assert data["title"] == "New Chat" - assert data["updated_at"] is not None - assert data["created_at"] is not None - assert len(self.chats.get_chats()) == 2 - - def test_get_user_chats(self): - self.test_get_session_user_chat_list() - - def test_get_user_archived_chats(self): - self.chats.archive_all_chats_by_user_id("2") - from open_webui.internal.db import Session - - Session.commit() - with mock_webui_user(id="2"): - response = self.fast_api_client.get(self.create_url("/all/archived")) - assert response.status_code == 200 - first_chat = response.json()[0] - assert first_chat["id"] is not None - assert first_chat["title"] == "New Chat" - assert first_chat["created_at"] is not None - assert first_chat["updated_at"] is not None - - def test_get_all_user_chats_in_db(self): - with mock_webui_user(id="4"): - response = self.fast_api_client.get(self.create_url("/all/db")) - assert response.status_code == 200 - assert len(response.json()) == 1 - - def test_get_archived_session_user_chat_list(self): - self.test_get_user_archived_chats() - - def test_archive_all_chats(self): - with mock_webui_user(id="2"): - response = self.fast_api_client.post(self.create_url("/archive/all")) - assert response.status_code == 200 - assert len(self.chats.get_archived_chats_by_user_id("2")) == 1 - - def test_get_shared_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - self.chats.update_chat_share_id_by_id(chat_id, chat_id) - with mock_webui_user(id="2"): - response = self.fast_api_client.get(self.create_url(f"/share/{chat_id}")) - assert response.status_code == 200 - data = response.json() - assert data["id"] == chat_id - assert data["chat"] == { - "name": "chat1", - "description": "chat1 description", - "tags": ["tag1", "tag2"], - "history": {"currentId": "1", "messages": []}, - } - assert data["id"] == chat_id - assert data["share_id"] == chat_id - assert data["title"] == "New Chat" - - def test_get_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - with mock_webui_user(id="2"): - response = self.fast_api_client.get(self.create_url(f"/{chat_id}")) - assert response.status_code == 200 - data = response.json() - assert data["id"] == chat_id - assert data["chat"] == { - "name": "chat1", - "description": "chat1 description", - "tags": ["tag1", "tag2"], - "history": {"currentId": "1", "messages": []}, - } - assert data["share_id"] is None - assert data["title"] == "New Chat" - assert data["user_id"] == "2" - - def test_update_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - with mock_webui_user(id="2"): - response = self.fast_api_client.post( - self.create_url(f"/{chat_id}"), - json={ - "chat": { - "name": "chat2", - "description": "chat2 description", - "tags": ["tag2", "tag4"], - "title": "Just another title", - } - }, - ) - assert response.status_code == 200 - data = response.json() - assert data["id"] == chat_id - assert data["chat"] == { - "name": "chat2", - "title": "Just another title", - "description": "chat2 description", - "tags": ["tag2", "tag4"], - "history": {"currentId": "1", "messages": []}, - } - assert data["share_id"] is None - assert data["title"] == "Just another title" - assert data["user_id"] == "2" - - def test_delete_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - with mock_webui_user(id="2"): - response = self.fast_api_client.delete(self.create_url(f"/{chat_id}")) - assert response.status_code == 200 - assert response.json() is True - - def test_clone_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - with mock_webui_user(id="2"): - response = self.fast_api_client.get(self.create_url(f"/{chat_id}/clone")) - - assert response.status_code == 200 - data = response.json() - assert data["id"] != chat_id - assert data["chat"] == { - "branchPointMessageId": "1", - "description": "chat1 description", - "history": {"currentId": "1", "messages": []}, - "name": "chat1", - "originalChatId": chat_id, - "tags": ["tag1", "tag2"], - "title": "Clone of New Chat", - } - assert data["share_id"] is None - assert data["title"] == "Clone of New Chat" - assert data["user_id"] == "2" - - def test_archive_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - with mock_webui_user(id="2"): - response = self.fast_api_client.get(self.create_url(f"/{chat_id}/archive")) - assert response.status_code == 200 - - chat = self.chats.get_chat_by_id(chat_id) - assert chat.archived is True - - def test_share_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - with mock_webui_user(id="2"): - response = self.fast_api_client.post(self.create_url(f"/{chat_id}/share")) - assert response.status_code == 200 - - chat = self.chats.get_chat_by_id(chat_id) - assert chat.share_id is not None - - def test_delete_shared_chat_by_id(self): - chat_id = self.chats.get_chats()[0].id - share_id = str(uuid.uuid4()) - self.chats.update_chat_share_id_by_id(chat_id, share_id) - with mock_webui_user(id="2"): - response = self.fast_api_client.delete(self.create_url(f"/{chat_id}/share")) - assert response.status_code - - chat = self.chats.get_chat_by_id(chat_id) - assert chat.share_id is None diff --git a/backend/open_webui/test/util/abstract_integration_test.py b/backend/open_webui/test/util/abstract_integration_test.py deleted file mode 100644 index e8492befb6..0000000000 --- a/backend/open_webui/test/util/abstract_integration_test.py +++ /dev/null @@ -1,161 +0,0 @@ -import logging -import os -import time - -import docker -import pytest -from docker import DockerClient -from pytest_docker.plugin import get_docker_ip -from fastapi.testclient import TestClient -from sqlalchemy import text, create_engine - - -log = logging.getLogger(__name__) - - -def get_fast_api_client(): - from main import app - - with TestClient(app) as c: - return c - - -class AbstractIntegrationTest: - BASE_PATH = None - - def create_url(self, path="", query_params=None): - if self.BASE_PATH is None: - raise Exception("BASE_PATH is not set") - parts = self.BASE_PATH.split("/") - parts = [part.strip() for part in parts if part.strip() != ""] - path_parts = path.split("/") - path_parts = [part.strip() for part in path_parts if part.strip() != ""] - query_parts = "" - if query_params: - query_parts = "&".join( - [f"{key}={value}" for key, value in query_params.items()] - ) - query_parts = f"?{query_parts}" - return "/".join(parts + path_parts) + query_parts - - @classmethod - def setup_class(cls): - pass - - def setup_method(self): - pass - - @classmethod - def teardown_class(cls): - pass - - def teardown_method(self): - pass - - -class AbstractPostgresTest(AbstractIntegrationTest): - DOCKER_CONTAINER_NAME = "postgres-test-container-will-get-deleted" - docker_client: DockerClient - - @classmethod - def _create_db_url(cls, env_vars_postgres: dict) -> str: - host = get_docker_ip() - user = env_vars_postgres["POSTGRES_USER"] - pw = env_vars_postgres["POSTGRES_PASSWORD"] - port = 8081 - db = env_vars_postgres["POSTGRES_DB"] - return f"postgresql://{user}:{pw}@{host}:{port}/{db}" - - @classmethod - def setup_class(cls): - super().setup_class() - try: - env_vars_postgres = { - "POSTGRES_USER": "user", - "POSTGRES_PASSWORD": "example", - "POSTGRES_DB": "openwebui", - } - cls.docker_client = docker.from_env() - cls.docker_client.containers.run( - "postgres:16.2", - detach=True, - environment=env_vars_postgres, - name=cls.DOCKER_CONTAINER_NAME, - ports={5432: ("0.0.0.0", 8081)}, - command="postgres -c log_statement=all", - ) - time.sleep(0.5) - - database_url = cls._create_db_url(env_vars_postgres) - os.environ["DATABASE_URL"] = database_url - retries = 10 - db = None - while retries > 0: - try: - from open_webui.config import OPEN_WEBUI_DIR - - db = create_engine(database_url, pool_pre_ping=True) - db = db.connect() - log.info("postgres is ready!") - break - except Exception as e: - log.warning(e) - time.sleep(3) - retries -= 1 - - if db: - # import must be after setting env! - cls.fast_api_client = get_fast_api_client() - db.close() - else: - raise Exception("Could not connect to Postgres") - except Exception as ex: - log.error(ex) - cls.teardown_class() - pytest.fail(f"Could not setup test environment: {ex}") - - def _check_db_connection(self): - from open_webui.internal.db import Session - - retries = 10 - while retries > 0: - try: - Session.execute(text("SELECT 1")) - Session.commit() - break - except Exception as e: - Session.rollback() - log.warning(e) - time.sleep(3) - retries -= 1 - - def setup_method(self): - super().setup_method() - self._check_db_connection() - - @classmethod - def teardown_class(cls) -> None: - super().teardown_class() - cls.docker_client.containers.get(cls.DOCKER_CONTAINER_NAME).remove(force=True) - - def teardown_method(self): - from open_webui.internal.db import Session - - # rollback everything not yet committed - Session.commit() - - # truncate all tables - tables = [ - "auth", - "chat", - "chatidtag", - "document", - "memory", - "model", - "prompt", - "tag", - '"user"', - ] - for table in tables: - Session.execute(text(f"TRUNCATE TABLE {table}")) - Session.commit()