office365manage/tests/test_app.py

from office365_admin import create_app
from office365_admin.settings import Settings


class FakeService:
    def list_licenses(self):
        return [{"skuId": "1", "skuPartNumber": "O365_BUSINESS", "availableUnits": 5, "consumedUnits": 5, "totalUnits": 10}]

    def list_users(self, search="", page=1, page_size=25):
        return {
            "items": [
                {
                    "id": "1",
                    "displayName": "Alice",
                    "userPrincipalName": "alice@example.com",
                    "mail": "alice@example.com",
                    "givenName": "Alice",
                    "surname": "Zhang",
                    "department": "IT",
                    "jobTitle": "Engineer",
                    "officeLocation": "",
                    "mobilePhone": "",
                    "usageLocation": "US",
                    "accountEnabled": True,
                    "assignedLicenses": ["1"],
                    "assignedLicensesCount": 1,
                    "licenseLabels": ["O365_BUSINESS"],
                    "createdDateTime": "",
                }
            ],
            "page": page,
            "pageSize": page_size,
            "total": 1,
            "totalBeforeSearch": 1,
            "summary": {"active": 1, "disabled": 0},
        }

    def list_user_identifiers(self, search=""):
        return {"identifiers": ["alice@example.com"], "total": 1}

    def get_user(self, identifier):
        return self.list_users()["items"][0]

    def create_user(self, payload):
        return {"user": self.get_user(payload["userPrincipalName"]), "temporaryPassword": "temp"}

    def update_user(self, identifier, payload):
        return {"user": self.get_user(identifier)}

    def delete_user(self, identifier):
        return {"user": self.get_user(identifier)}

    def reset_password(self, identifier, payload=None):
        return {"user": self.get_user(identifier), "temporaryPassword": "newpass"}

    def batch_create(self, rows, progress_callback=None):
        return {"operation": "create", "total": len(rows), "successCount": len(rows), "failureCount": 0, "results": []}

    def batch_update(self, rows, progress_callback=None):
        return {"operation": "update", "total": len(rows), "successCount": len(rows), "failureCount": 0, "results": []}

    def batch_delete(self, identifiers, progress_callback=None):
        return {"operation": "delete", "total": len(identifiers), "successCount": len(identifiers), "failureCount": 0, "results": []}

    def batch_reset_password(self, rows, progress_callback=None):
        return {"operation": "reset-password", "total": len(rows), "successCount": len(rows), "failureCount": 0, "results": []}


class FakeTaskManager:
    def __init__(self):
        self.last_task = {
            "id": "task-1",
            "operation": "reset-password",
            "status": "succeeded",
            "message": "任务完成，成功 1，失败 0",
            "createdAt": "2026-03-21T12:00:00",
            "startedAt": "2026-03-21T12:00:00",
            "finishedAt": "2026-03-21T12:00:01",
            "total": 1,
            "completed": 1,
            "successCount": 1,
            "failureCount": 0,
            "progressPercent": 100,
            "currentItem": "alice@example.com",
            "currentMessage": "任务完成，成功 1，失败 0",
            "recentFailures": [],
        }

    def submit(self, operation, total, runner):
        self.last_task = {
            **self.last_task,
            "operation": operation,
            "total": total,
            "message": f"{operation} submitted",
        }
        return self.last_task

    def get_task(self, task_id):
        return self.last_task


def build_settings():
    return Settings(
        app_name="Test App",
        host="127.0.0.1",
        port=8000,
        debug=False,
        session_secret="test-secret",
        auth_enabled=True,
        admin_username="admin",
        admin_password="password",
        client_id="client",
        tenant_id="tenant",
        client_secret="secret",
        default_password="pass",
        default_domain="example.com",
        default_usage_location="US",
        default_license_sku="O365_BUSINESS",
        force_change_password=True,
        graph_base_url="https://example.test",
        token_endpoint="https://login.example.test/token",
        scope="scope",
    )


def build_client():
    app = create_app(
        settings_override=build_settings(),
        service_factory=FakeService(),
        task_manager_factory=FakeTaskManager(),
    )
    app.config["TESTING"] = True
    return app.test_client()


def login(client):
    response = client.post("/api/login", json={"username": "admin", "password": "password"})
    assert response.status_code == 200


def test_users_requires_login():
    client = build_client()
    response = client.get("/api/users")
    assert response.status_code == 401


def test_login_and_list_users():
    client = build_client()
    login(client)
    response = client.get("/api/users")
    assert response.status_code == 200
    payload = response.get_json()
    assert payload["success"] is True
    assert payload["data"]["items"][0]["userPrincipalName"] == "alice@example.com"


def test_list_user_identifiers_after_login():
    client = build_client()
    login(client)
    response = client.get("/api/users/selection")
    assert response.status_code == 200
    payload = response.get_json()
    assert payload["success"] is True
    assert payload["data"]["identifiers"] == ["alice@example.com"]


def test_create_user_after_login():
    client = build_client()
    login(client)
    response = client.post("/api/users", json={"userPrincipalName": "alice@example.com"})
    assert response.status_code == 201
    payload = response.get_json()
    assert payload["success"] is True
    assert payload["data"]["temporaryPassword"] == "temp"


def test_batch_reset_password_submits_task():
    client = build_client()
    login(client)
    response = client.post("/api/users/batch/reset-password", json={"rows": [{"userPrincipalName": "alice@example.com"}]})
    assert response.status_code == 202
    payload = response.get_json()
    assert payload["success"] is True
    assert payload["data"]["id"] == "task-1"