206 lines
6.6 KiB
Python
206 lines
6.6 KiB
Python
"""Tests for U7: bitable CLI subcommands.
|
|
|
|
Requires PostgreSQL — marked ``postgres``. Uses Typer's CliRunner.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from agentkit.cli.bitable import bitable_app
|
|
|
|
pytestmark = pytest.mark.postgres
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def db_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Ensure DATABASE_URL is set for CLI tests."""
|
|
url = os.environ.get("DATABASE_URL") or os.environ.get("AGENTKIT_DATABASE_URL")
|
|
if not url:
|
|
pytest.skip("PostgreSQL not available (set DATABASE_URL)")
|
|
monkeypatch.setenv("DATABASE_URL", url)
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_schema(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Drop and recreate bitable schema before each test."""
|
|
import asyncio
|
|
|
|
url = os.environ.get("DATABASE_URL") or os.environ.get("AGENTKIT_DATABASE_URL")
|
|
if not url:
|
|
pytest.skip("PostgreSQL not available")
|
|
|
|
from agentkit.bitable.db import BitableDB
|
|
from sqlalchemy import text
|
|
|
|
async def _clean():
|
|
db = BitableDB()
|
|
await db.init()
|
|
async with db.engine.begin() as conn:
|
|
await conn.execute(text("DROP SCHEMA IF EXISTS bitable CASCADE"))
|
|
await db.init()
|
|
await db.close()
|
|
|
|
asyncio.run(_clean())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list-tables
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_tables_empty(db_env, clean_schema) -> None:
|
|
result = runner.invoke(bitable_app, ["list-tables"])
|
|
assert result.exit_code == 0
|
|
assert "No tables found" in result.output
|
|
|
|
|
|
def test_list_tables_after_create(db_env, clean_schema) -> None:
|
|
# Create a table first
|
|
runner.invoke(bitable_app, ["create-table", "--name", "TestTable"])
|
|
result = runner.invoke(bitable_app, ["list-tables"])
|
|
assert result.exit_code == 0
|
|
assert "TestTable" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create-table
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_table_success(db_env, clean_schema) -> None:
|
|
result = runner.invoke(
|
|
bitable_app,
|
|
["create-table", "--name", "MyTable", "--description", "A test table"],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "Created table" in result.output
|
|
assert "MyTable" in result.output
|
|
assert "A test table" in result.output
|
|
|
|
|
|
def test_create_table_minimal(db_env, clean_schema) -> None:
|
|
result = runner.invoke(bitable_app, ["create-table", "--name", "Minimal"])
|
|
assert result.exit_code == 0
|
|
assert "Minimal" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# query
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_query_table_not_found(db_env, clean_schema) -> None:
|
|
result = runner.invoke(bitable_app, ["query", "--table", "nonexistent-id"])
|
|
assert result.exit_code == 1
|
|
assert "not found" in result.output
|
|
|
|
|
|
def test_query_empty_table(db_env, clean_schema) -> None:
|
|
# Create a table first
|
|
create_result = runner.invoke(bitable_app, ["create-table", "--name", "Empty"])
|
|
assert create_result.exit_code == 0
|
|
# Extract table ID from output — it's on the "ID:" line
|
|
lines = create_result.output.split("\n")
|
|
table_id = None
|
|
for line in lines:
|
|
if "ID:" in line:
|
|
# Extract the cyan-colored ID
|
|
table_id = line.split("ID:")[1].strip().strip("[]").split(" ")[0]
|
|
# Remove rich formatting
|
|
table_id = table_id.replace("[cyan]", "").replace("[/cyan]", "")
|
|
break
|
|
assert table_id is not None, f"Could not extract table ID from: {create_result.output}"
|
|
|
|
result = runner.invoke(bitable_app, ["query", "--table", table_id])
|
|
assert result.exit_code == 0
|
|
assert "No records found" in result.output
|
|
|
|
|
|
def test_query_with_records(db_env, clean_schema) -> None:
|
|
"""Create table + field + records via service, then query via CLI."""
|
|
import asyncio
|
|
|
|
from agentkit.bitable.db import BitableDB
|
|
from agentkit.bitable.models import FieldType
|
|
from agentkit.bitable.service import BitableService
|
|
|
|
async def _setup():
|
|
db = BitableDB()
|
|
await db.init()
|
|
service = BitableService(db)
|
|
table = await service.create_table(name="Data")
|
|
field = await service.create_field(
|
|
table_id=table.id, name="name", field_type=FieldType.text
|
|
)
|
|
await service.create_record(table_id=table.id, values={field.id: "Alice"})
|
|
await service.create_record(table_id=table.id, values={field.id: "Bob"})
|
|
await db.close()
|
|
return table.id
|
|
|
|
table_id = asyncio.run(_setup())
|
|
|
|
result = runner.invoke(bitable_app, ["query", "--table", table_id, "--limit", "10"])
|
|
assert result.exit_code == 0
|
|
assert "Alice" in result.output
|
|
assert "Bob" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# import-excel
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_import_excel_file_not_found(db_env, clean_schema) -> None:
|
|
result = runner.invoke(
|
|
bitable_app,
|
|
["import-excel", "--file", "/nonexistent/file.xlsx"],
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "not found" in result.output
|
|
|
|
|
|
def test_import_excel_success(db_env, clean_schema, tmp_path: Path) -> None:
|
|
"""Create a real xlsx file and import it."""
|
|
from openpyxl import Workbook
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Sheet1"
|
|
ws.append(["name", "age"])
|
|
ws.append(["Alice", 30])
|
|
ws.append(["Bob", 25])
|
|
xlsx_path = tmp_path / "test.xlsx"
|
|
wb.save(xlsx_path)
|
|
|
|
result = runner.invoke(
|
|
bitable_app,
|
|
["import-excel", "--file", str(xlsx_path), "--table", "Imported"],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "Created table" in result.output
|
|
assert "Imported" in result.output
|
|
assert "Imported 2 records" in result.output
|
|
assert "name" in result.output
|
|
assert "age" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error path: no DATABASE_URL
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_no_database_url(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""CLI should exit with clear error when DATABASE_URL is not set."""
|
|
monkeypatch.delenv("DATABASE_URL", raising=False)
|
|
monkeypatch.delenv("AGENTKIT_DATABASE_URL", raising=False)
|
|
result = runner.invoke(bitable_app, ["list-tables"])
|
|
assert result.exit_code == 1
|
|
assert "DATABASE_URL" in result.output
|