shanxia2.0/backend/tests/test_post_users_omission.py

"""Real position-user PUT route against SQLite; never connect to business databases."""
from datetime import datetime

import pymysql
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app.api import admin
from app.models.models import SysPost, SysUser, SysUserPost


@pytest.fixture
def context(monkeypatch):
    def forbidden(*args, **kwargs):
        raise AssertionError('Business database access is forbidden')
    monkeypatch.setattr(pymysql, 'connect', forbidden)
    monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
    engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
    for model in (SysPost, SysUser, SysUserPost):
        model.__table__.create(engine)
    db = Session(engine, expire_on_commit=False)
    for uid, tenant, deleted in [('actor', 't', 0), ('u1', 't', 0), ('u2', 't', 0), ('foreign', 'other', 0), ('deleted', 't', 1)]:
        db.add(SysUser(id=uid, user_no=uid, nick_name=uid, tenant_no=tenant, del_flag=deleted))
    for pid, tenant, deleted in [('p', 't', 0), ('p2', 't', 0), ('foreign-p', 'other', 0), ('deleted-p', 't', 1)]:
        db.add(SysPost(id=pid, post_name=pid, post_code=pid, tenant_no=tenant, del_flag=deleted))
    for rid, uid, pid, tenant in [('kept', 'u1', 'p', 't'), ('other-post', 'u1', 'p2', 't'), ('other-tenant', 'foreign', 'foreign-p', 'other')]:
        db.add(SysUserPost(id=rid, user_id=uid, post_id=pid, tenant_no=tenant, del_flag=0,
                           create_by='creator', create_time=datetime(2020, 1, 1)))
    db.commit()
    writes, commits = [], []
    def capture(_conn, _cursor, statement, *_args):
        if statement.lstrip().upper().startswith(('INSERT', 'UPDATE', 'DELETE')):
            writes.append(statement)
    event.listen(engine, 'before_cursor_execute', capture)
    event.listen(db, 'after_commit', lambda _db: commits.append(True))
    app = FastAPI()
    app.add_api_route('/posts/{post_id}/users', admin.update_post_bound_users, methods=['PUT'])
    app.dependency_overrides[admin.get_db] = lambda: db
    app.dependency_overrides[admin.require_admin_or_super_admin] = lambda: {'user_id': 'actor'}
    with TestClient(app) as client:
        yield client, db, writes, commits
    db.close()
    engine.dispose()


def put(context, payload, post_id='p'):
    return context[0].put(f'/posts/{post_id}/users', json=payload)


def test_omitted_user_ids_does_not_write_or_commit(context):
    response = put(context, {})
    assert response.status_code == 200
    _, db, writes, commits = context
    assert db.get(SysUserPost, 'kept').del_flag == 0
    assert not writes and not commits
    assert not db.dirty and not db.new


@pytest.mark.parametrize('key', ['userIds', 'user_ids'])
def test_explicit_empty_array_clears_only_current_post(context, key):
    assert put(context, {key: []}).status_code == 200
    db = context[1]
    assert db.get(SysUserPost, 'kept').del_flag == 1
    assert db.get(SysUserPost, 'other-post').del_flag == 0
    assert db.get(SysUserPost, 'other-tenant').del_flag == 0
    assert db.query(SysUserPost).count() == 3


@pytest.mark.parametrize('key', ['userIds', 'user_ids'])
def test_explicit_null_is_rejected_before_any_write(context, key):
    assert put(context, {key: None}).status_code == 422
    assert context[1].get(SysUserPost, 'kept').del_flag == 0
    assert not context[2] and not context[3]


def test_nonempty_array_preserves_old_identity_and_adds_only_new_user(context):
    for _ in range(2):
        assert put(context, {'userIds': ['u1', 'u2']}).status_code == 200
    db = context[1]
    kept = db.get(SysUserPost, 'kept')
    assert kept.del_flag == 0 and kept.create_by == 'creator'
    assert kept.create_time == datetime(2020, 1, 1)
    assert db.query(SysUserPost).filter_by(post_id='p', del_flag=0).count() == 2
    assert db.query(SysUserPost).count() == 4
    assert db.get(SysUserPost, 'other-post').del_flag == 0
    assert db.get(SysUserPost, 'other-tenant').del_flag == 0


@pytest.mark.parametrize('uid', ['foreign', 'missing', 'deleted'])
def test_invalid_user_rejected_before_mutating_existing_bindings(context, uid):
    response = put(context, {'userIds': ['u2', uid]})
    assert response.json()['code'] == 400
    assert context[1].get(SysUserPost, 'kept').del_flag == 0
    assert not context[2] and not context[3]


@pytest.mark.parametrize('post_id', ['foreign-p', 'missing', 'deleted-p'])
@pytest.mark.parametrize('payload', [{}, {'userIds': []}])
def test_post_scope_is_validated_even_when_user_ids_are_omitted(context, post_id, payload):
    response = put(context, payload, post_id)
    assert response.json()['code'] == 404
    assert not context[2] and not context[3]