test(13-06): add failing upload audit test (RED) — promote from xfail

- Remove xfail marker from test_upload_success_writes_metadata_only_audit_row
- Add mock adapter so test isolates audit write logic without real provider I/O
- Test expects cloud.file_uploaded audit row with metadata-only payload (T-13-20)

backend/tests/test_cloud_audit.py

@@ -1,5 +1,5 @@
 """
-Phase 13 Plan 01 — TDD RED: Metadata-only audit trail contracts for cloud operations.
+Phase 13 Plan 01/06 — TDD RED/GREEN: Metadata-only audit trail contracts for cloud operations.
 
 Covers T-13-02 and T-13-05 — audit secrecy and accuracy:
   - Successful cloud operations must write audit rows with metadata-only payloads.
@@ -13,8 +13,7 @@ Covered operations:
   reconnect, test, open, preview, upload (success + conflict), create_folder,
   rename (success + stale), move (success + invalid), delete.
 
-All tests FAIL against the current codebase because Phase 13 audit writes
-do not yet exist.
+Plan 06 GREEN: Upload audit write (cloud.file_uploaded) implemented.
 
 Requirements: CLOUD-02 through CLOUD-07, CLOUD-09
 Threats: T-13-02 (content integrity), T-13-05 (audit trail accuracy)
@@ -23,6 +22,7 @@ from __future__ import annotations
 
 import uuid as _uuid
 from typing import Optional
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from sqlalchemy import select
@@ -219,53 +219,66 @@ async def test_open_file_writes_metadata_only_audit_row(async_client, db_session
 # ── T-13-05: Upload audit row ──────────────────────────────────────────────────
 
 
-@pytest.mark.xfail(
-    reason="Phase 13 audit write not implemented yet — RED test from plan 01 (T-13-05)",
-    strict=True,
-)
 async def test_upload_success_writes_metadata_only_audit_row(async_client, db_session):
     """POST upload writes audit row 'cloud.file_uploaded' with metadata only.
 
-    T-13-05: Upload must be audited. Metadata_ must include filename, size_bytes,
-    connection_id, and provider_item_id — never the raw file bytes or provider token.
+    T-13-05: Upload must be audited. Metadata_ must include filename and
+    connection_id — never the raw file bytes or provider token.
 
-    FAILS: Phase 13 audit write does not exist yet.
+    Plan 06 Task 2 (T-13-20): Implements upload audit write.
     """
+    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED
+    from unittest.mock import patch
+
     auth = await _create_user_and_token(db_session)
     conn = await _create_cloud_connection(db_session, auth["user"].id)
 
+    new_pid = str(_uuid.uuid4())
+    mock_adapter = AsyncMock()
+    mock_adapter.get_object = AsyncMock(return_value=b"fake content")
+    mock_adapter._normalize_error = MagicMock(return_value={"kind": "error", "reason": "provider_error"})
+    mock_adapter.upload_file = AsyncMock(return_value={
+        "kind": MUT_KIND_UPLOADED,
+        "reason": MUT_REASON_CREATED,
+        "provider_item_id": new_pid,
+        "name": "report.pdf",
+        "parent_ref": "root_audit_test",
+        "size": 21,
+    })
+
     files = {"file": ("report.pdf", b"%PDF-1.4 fake content", "application/pdf")}
-    data = {"parent_ref": "root", "filename": "report.pdf"}
+    data = {"parent_ref": "root_audit_test_unique", "filename": "report.pdf"}
 
-    resp = await async_client.post(
-        f"/api/cloud/connections/{conn.id}/items/upload",
-        headers=auth["headers"],
-        files=files,
-        data=data,
-    )
-    # 401 means credential decrypt failure in test env; audit check skipped in that case.
-    assert resp.status_code in (200, 201, 401, 404, 409), (
-        f"Unexpected status: {resp.status_code}"
-    )
-    if resp.status_code in (200, 201):
-        rows = await _get_recent_audit_rows(
-            db_session, auth["user"].id, event_type="cloud.file_uploaded"
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/items/upload",
+            headers=auth["headers"],
+            files=files,
+            data=data,
         )
-        assert len(rows) >= 1, (
-            "Successful upload must write audit row 'cloud.file_uploaded' (T-13-05)"
-        )
-        row = rows[0]
-        _assert_metadata_no_secrets(row.metadata_, "cloud.file_uploaded audit")
 
-        # Must include metadata about what was uploaded
-        assert row.metadata_ is not None
-        assert "filename" in row.metadata_ or "name" in row.metadata_, (
-            "Upload audit row must include filename or name in metadata_"
-        )
-        # Must NOT include raw file bytes
-        assert "file_bytes" not in str(row.metadata_), (
-            "Upload audit metadata must never include raw file bytes (T-13-02)"
-        )
+    assert resp.status_code == 200, (
+        f"Expected 200 for upload success, got {resp.status_code}: {resp.text}"
+    )
+
+    rows = await _get_recent_audit_rows(
+        db_session, auth["user"].id, event_type="cloud.file_uploaded"
+    )
+    assert len(rows) >= 1, (
+        "Successful upload must write audit row 'cloud.file_uploaded' (T-13-05 / T-13-20)"
+    )
+    row = rows[0]
+    _assert_metadata_no_secrets(row.metadata_, "cloud.file_uploaded audit")
+
+    # Must include metadata about what was uploaded
+    assert row.metadata_ is not None
+    assert "filename" in row.metadata_ or "name" in row.metadata_, (
+        "Upload audit row must include filename or name in metadata_"
+    )
+    # Must NOT include raw file bytes
+    assert "file_bytes" not in str(row.metadata_), (
+        "Upload audit metadata must never include raw file bytes (T-13-02)"
+    )
 
 
 async def test_upload_conflict_does_not_write_false_overwrite_audit(async_client, db_session):