|
| 1 | +from pathlib import Path |
| 2 | +from typing import Any, cast |
| 3 | + |
| 4 | +import httpx |
| 5 | +import pytest |
| 6 | +from openpyxl import Workbook |
| 7 | + |
| 8 | +from excel_dbapi import connect |
| 9 | +from excel_dbapi.connection import ExcelConnection |
| 10 | +from excel_dbapi.exceptions import Error, OperationalError, ProgrammingError |
| 11 | + |
| 12 | + |
| 13 | +def _create_workbook( |
| 14 | + path: Path, sheet: str, headers: list[object], rows: list[list[object]] |
| 15 | +) -> None: |
| 16 | + workbook = Workbook() |
| 17 | + worksheet = workbook.active |
| 18 | + assert worksheet is not None |
| 19 | + worksheet.title = sheet |
| 20 | + worksheet.append(headers) |
| 21 | + for row in rows: |
| 22 | + worksheet.append(row) |
| 23 | + workbook.save(path) |
| 24 | + workbook.close() |
| 25 | + |
| 26 | + |
| 27 | +def test_graph_invalid_credential_is_wrapped_as_operational_error() -> None: |
| 28 | + with pytest.raises(OperationalError, match="Cannot normalise"): |
| 29 | + connect( |
| 30 | + "msgraph://drives/d/items/i", |
| 31 | + engine="graph", |
| 32 | + credential=cast(Any, 42), |
| 33 | + ) |
| 34 | + |
| 35 | + |
| 36 | +def test_graph_token_provider_failure_is_translated_during_execute() -> None: |
| 37 | + class ExplodingTokenProvider: |
| 38 | + def get_token(self, *args: Any) -> Any: |
| 39 | + del args |
| 40 | + raise RuntimeError("token boom") |
| 41 | + |
| 42 | + transport = httpx.MockTransport(lambda request: httpx.Response(200, json={})) |
| 43 | + with ExcelConnection( |
| 44 | + "msgraph://drives/drv-test/items/itm-test", |
| 45 | + engine="graph", |
| 46 | + credential=ExplodingTokenProvider(), |
| 47 | + transport=transport, |
| 48 | + ) as conn: |
| 49 | + cursor = conn.cursor() |
| 50 | + with pytest.raises( |
| 51 | + Error, match="Failed to acquire authentication token: token boom" |
| 52 | + ): |
| 53 | + cursor.execute("SELECT * FROM Employees") |
| 54 | + |
| 55 | + |
| 56 | +def test_executor_resolves_unicode_sheet_names_with_casefold(tmp_path: Path) -> None: |
| 57 | + file_path = tmp_path / "unicode-sheet-casefold.xlsx" |
| 58 | + _create_workbook(file_path, "Straße", ["id"], [[1]]) |
| 59 | + |
| 60 | + with ExcelConnection(str(file_path), engine="openpyxl") as conn: |
| 61 | + cursor = conn.cursor() |
| 62 | + cursor.execute("SELECT * FROM STRASSE") |
| 63 | + assert cursor.fetchall() == [(1,)] |
| 64 | + |
| 65 | + |
| 66 | +def test_lastrowid_is_cleared_after_failed_execute(tmp_path: Path) -> None: |
| 67 | + file_path = tmp_path / "lastrowid-reset.xlsx" |
| 68 | + _create_workbook(file_path, "Sheet1", ["id", "name"], [[1, "Alice"]]) |
| 69 | + |
| 70 | + with ExcelConnection(str(file_path), engine="openpyxl") as conn: |
| 71 | + cursor = conn.cursor() |
| 72 | + cursor.execute("INSERT INTO Sheet1 VALUES (2, 'Bob')") |
| 73 | + assert cursor.lastrowid is not None |
| 74 | + |
| 75 | + with pytest.raises(ProgrammingError): |
| 76 | + cursor.execute("SELECT * FROM MissingSheet") |
| 77 | + |
| 78 | + assert cursor.lastrowid is None |
0 commit comments