Initial commit: Office365 web management platform

tests/test_app.py

@@ -0,0 +1,184 @@
+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"

tests/test_batch.py

@@ -0,0 +1,31 @@
+from office365_admin.batch import parse_identifier_content, parse_table_content
+
+
+def test_parse_table_content_from_csv():
+    content = "userPrincipalName,displayName,department\nalice, Alice Zhang,Sales\n"
+    rows = parse_table_content(content)
+    assert rows == [
+        {
+            "userPrincipalName": "alice",
+            "displayName": "Alice Zhang",
+            "department": "Sales",
+        }
+    ]
+
+
+def test_parse_table_content_from_json():
+    rows = parse_table_content('[{"userPrincipalName":"bob@example.com","department":"IT"}]')
+    assert rows[0]["userPrincipalName"] == "bob@example.com"
+    assert rows[0]["department"] == "IT"
+
+
+def test_parse_identifier_content_from_lines():
+    values = parse_identifier_content("alice\nbob@example.com\n")
+    assert values == ["alice", "bob@example.com"]
+
+
+def test_parse_identifier_content_from_csv():
+    content = "userPrincipalName,displayName\nalice@example.com,Alice\n"
+    values = parse_identifier_content(content)
+    assert values == ["alice@example.com"]
+