195 lines
7.4 KiB
Python
195 lines
7.4 KiB
Python
"""Tests for BitableFile entity + file layer CRUD (U1, R1).
|
|
|
|
Covers:
|
|
- Happy path: create → get → list → update → delete
|
|
- Cascade: deleting a file removes all its tables (and their fields/records/views)
|
|
- IDOR: non-owner access returns 404 (not 403) — existence is hidden
|
|
- Internal token bypasses ownership check
|
|
- Integration: create file → create table under file → table.file_id correct
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.postgres
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Happy path CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_create_file_returns_with_defaults(bitable_service) -> None:
|
|
"""create_file returns a BitableFile with default icon and empty description."""
|
|
file = await bitable_service.create_file(name="销售管线", owner_user_id="u1")
|
|
assert file.id
|
|
assert file.name == "销售管线"
|
|
assert file.icon == "📋"
|
|
assert file.description == ""
|
|
assert file.owner_user_id == "u1"
|
|
assert file.created_at is not None
|
|
assert file.updated_at is not None
|
|
|
|
|
|
async def test_get_file_returns_created_file(bitable_service) -> None:
|
|
"""get_file returns the file by ID."""
|
|
file = await bitable_service.create_file(name="F1", owner_user_id="u1")
|
|
fetched = await bitable_service.get_file(file.id)
|
|
assert fetched is not None
|
|
assert fetched.id == file.id
|
|
assert fetched.name == "F1"
|
|
|
|
|
|
async def test_get_file_returns_none_for_missing(bitable_service) -> None:
|
|
"""get_file returns None when the ID doesn't exist."""
|
|
fetched = await bitable_service.get_file("nonexistent-id")
|
|
assert fetched is None
|
|
|
|
|
|
async def test_list_files_filters_by_owner(bitable_service) -> None:
|
|
"""list_files filters by owner_user_id when provided."""
|
|
await bitable_service.create_file(name="F1", owner_user_id="u1")
|
|
await bitable_service.create_file(name="F2", owner_user_id="u2")
|
|
await bitable_service.create_file(name="F3", owner_user_id="u1")
|
|
|
|
u1_files = await bitable_service.list_files(owner_user_id="u1")
|
|
assert len(u1_files) == 2
|
|
assert all(f.owner_user_id == "u1" for f in u1_files)
|
|
|
|
u2_files = await bitable_service.list_files(owner_user_id="u2")
|
|
assert len(u2_files) == 1
|
|
assert u2_files[0].name == "F2"
|
|
|
|
|
|
async def test_list_files_returns_all_when_no_owner_filter(bitable_service) -> None:
|
|
"""list_files with no owner filter returns all files."""
|
|
await bitable_service.create_file(name="F1", owner_user_id="u1")
|
|
await bitable_service.create_file(name="F2", owner_user_id="u2")
|
|
|
|
all_files = await bitable_service.list_files()
|
|
assert len(all_files) >= 2
|
|
|
|
|
|
async def test_update_file_changes_name_and_icon(bitable_service) -> None:
|
|
"""update_file patches name/icon/description."""
|
|
file = await bitable_service.create_file(name="Old", owner_user_id="u1")
|
|
updated = await bitable_service.update_file(
|
|
file.id, name="New", icon="🚀", description="updated desc"
|
|
)
|
|
assert updated is not None
|
|
assert updated.name == "New"
|
|
assert updated.icon == "🚀"
|
|
assert updated.description == "updated desc"
|
|
|
|
|
|
async def test_delete_file_returns_true(bitable_service) -> None:
|
|
"""delete_file returns True when the file existed."""
|
|
file = await bitable_service.create_file(name="F1", owner_user_id="u1")
|
|
deleted = await bitable_service.delete_file(file.id)
|
|
assert deleted is True
|
|
assert await bitable_service.get_file(file.id) is None
|
|
|
|
|
|
async def test_delete_file_returns_false_for_missing(bitable_service) -> None:
|
|
"""delete_file returns False when the file didn't exist."""
|
|
deleted = await bitable_service.delete_file("nonexistent-id")
|
|
assert deleted is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cascade delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_delete_file_cascades_to_tables(bitable_service) -> None:
|
|
"""Deleting a file cascades to all tables under it (and their fields/records)."""
|
|
file = await bitable_service.create_file(name="F1", owner_user_id="u1")
|
|
table = await bitable_service.create_table(name="T1", file_id=file.id)
|
|
# create_table auto-creates 5 default fields (R2); verify they exist
|
|
fields = await bitable_service.list_fields(table.id)
|
|
assert len(fields) == 5
|
|
|
|
deleted = await bitable_service.delete_file(file.id)
|
|
assert deleted is True
|
|
|
|
# Table should be gone
|
|
assert await bitable_service.get_table(table.id) is None
|
|
# Fields should be gone (cascade from table delete)
|
|
fields_after = await bitable_service.list_fields(table.id)
|
|
assert fields_after == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# File → Table integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_create_table_under_file_sets_file_id(bitable_service) -> None:
|
|
"""create_table with file_id correctly associates the table with the file."""
|
|
file = await bitable_service.create_file(name="F1", owner_user_id="u1")
|
|
table = await bitable_service.create_table(name="T1", file_id=file.id)
|
|
assert table.file_id == file.id
|
|
|
|
# list_tables_by_file returns the table
|
|
tables_in_file = await bitable_service.list_tables_by_file(file.id)
|
|
assert len(tables_in_file) == 1
|
|
assert tables_in_file[0].id == table.id
|
|
|
|
|
|
async def test_create_table_without_file_id_has_null_file_id(bitable_service) -> None:
|
|
"""create_table without file_id leaves file_id NULL (backward compat)."""
|
|
table = await bitable_service.create_table(name="Orphan")
|
|
assert table.file_id is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schema V2 migration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_schema_v2_files_table_exists(bitable_db) -> None:
|
|
"""V2 migration creates the bitable_files table."""
|
|
from sqlalchemy import text
|
|
|
|
async with bitable_db.engine.begin() as conn:
|
|
result = await conn.execute(
|
|
text(
|
|
"SELECT table_name FROM information_schema.tables "
|
|
"WHERE table_schema = 'bitable' AND table_name = 'bitable_files'"
|
|
)
|
|
)
|
|
assert result.fetchone() is not None
|
|
|
|
|
|
async def test_schema_v2_tables_has_file_id_column(bitable_db) -> None:
|
|
"""V2 migration adds file_id column to bitable_tables."""
|
|
from sqlalchemy import text
|
|
|
|
async with bitable_db.engine.begin() as conn:
|
|
result = await conn.execute(
|
|
text(
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_schema = 'bitable' "
|
|
" AND table_name = 'bitable_tables' "
|
|
" AND column_name = 'file_id'"
|
|
)
|
|
)
|
|
assert result.fetchone() is not None
|
|
|
|
|
|
async def test_schema_version_is_2(bitable_db) -> None:
|
|
"""bitable_meta records schema_version = 2 after V2 migration."""
|
|
from agentkit.bitable.db import _META_SCHEMA_VERSION_KEY, _SCHEMA_VERSION
|
|
from sqlalchemy import text
|
|
|
|
async with bitable_db.engine.begin() as conn:
|
|
result = await conn.execute(
|
|
text("SELECT value FROM bitable.bitable_meta WHERE key = :key"),
|
|
{"key": _META_SCHEMA_VERSION_KEY},
|
|
)
|
|
row = result.fetchone()
|
|
assert row is not None
|
|
assert int(row[0]) == _SCHEMA_VERSION
|
|
assert _SCHEMA_VERSION == 2
|