repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | from unittest.mock import MagicMock |
| #2 | |
| #3 | import psycopg |
| #4 | import pytest |
| #5 | |
| #6 | from embedchain.loaders.postgres import PostgresLoader |
| #7 | |
| #8 | |
| #9 | @pytest.fixture |
| #10 | def postgres_loader(mocker): |
| #11 | with mocker.patch.object(psycopg, "connect"): |
| #12 | config = {"url": "postgres://user:password@localhost:5432/database"} |
| #13 | loader = PostgresLoader(config=config) |
| #14 | yield loader |
| #15 | |
| #16 | |
| #17 | def test_postgres_loader_initialization(postgres_loader): |
| #18 | assert postgres_loader.connection is not None |
| #19 | assert postgres_loader.cursor is not None |
| #20 | |
| #21 | |
| #22 | def test_postgres_loader_invalid_config(): |
| #23 | with pytest.raises(ValueError, match="Must provide the valid config. Received: None"): |
| #24 | PostgresLoader(config=None) |
| #25 | |
| #26 | |
| #27 | def test_load_data(postgres_loader, monkeypatch): |
| #28 | mock_cursor = MagicMock() |
| #29 | monkeypatch.setattr(postgres_loader, "cursor", mock_cursor) |
| #30 | |
| #31 | query = "SELECT * FROM table" |
| #32 | mock_cursor.fetchall.return_value = [(1, "data1"), (2, "data2")] |
| #33 | |
| #34 | result = postgres_loader.load_data(query) |
| #35 | |
| #36 | assert "doc_id" in result |
| #37 | assert "data" in result |
| #38 | assert len(result["data"]) == 2 |
| #39 | assert result["data"][0]["meta_data"]["url"] == query |
| #40 | assert result["data"][1]["meta_data"]["url"] == query |
| #41 | assert mock_cursor.execute.called_with(query) |
| #42 | |
| #43 | |
| #44 | def test_load_data_exception(postgres_loader, monkeypatch): |
| #45 | mock_cursor = MagicMock() |
| #46 | monkeypatch.setattr(postgres_loader, "cursor", mock_cursor) |
| #47 | |
| #48 | _ = "SELECT * FROM table" |
| #49 | mock_cursor.execute.side_effect = Exception("Mocked exception") |
| #50 | |
| #51 | with pytest.raises( |
| #52 | ValueError, match=r"Failed to load data using query=SELECT \* FROM table with: Mocked exception" |
| #53 | ): |
| #54 | postgres_loader.load_data("SELECT * FROM table") |
| #55 | |
| #56 | |
| #57 | def test_close_connection(postgres_loader): |
| #58 | postgres_loader.close_connection() |
| #59 | assert postgres_loader.cursor is None |
| #60 | assert postgres_loader.connection is None |
| #61 |