130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""Квота личного пространства OpenCloud (поле quota в /graph/v1.0/me/drives).
|
||
|
||
Наличие поля на нашем сервере не проверено: без него (или при сбое запроса)
|
||
интерфейс просто ничего не показывает.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import Mock, patch
|
||
|
||
import requests
|
||
|
||
import main
|
||
from app.opencloud_access import (
|
||
DriveQuota,
|
||
OpenCloudUploadError,
|
||
resolve_personal_drive,
|
||
resolve_personal_drive_id,
|
||
)
|
||
|
||
|
||
def _drives_response(personal: dict) -> Mock:
|
||
return Mock(
|
||
status_code=200,
|
||
json=lambda: {
|
||
"value": [
|
||
{"id": "s$shared", "driveType": "project", "quota": {"used": 1}},
|
||
{"id": "s$personal", "driveType": "personal", **personal},
|
||
]
|
||
},
|
||
)
|
||
|
||
|
||
def test_quota_is_returned_with_drive_id() -> None:
|
||
response = _drives_response(
|
||
{
|
||
"quota": {
|
||
"used": 3 * 1024**3,
|
||
"total": 10 * 1024**3,
|
||
"remaining": 7 * 1024**3,
|
||
"state": "normal",
|
||
}
|
||
}
|
||
)
|
||
with patch("app.opencloud_access.requests.get", return_value=response):
|
||
drive = resolve_personal_drive("raykovmse", "token")
|
||
|
||
assert drive.id == "s$personal"
|
||
assert drive.quota == DriveQuota(
|
||
used=3 * 1024**3, total=10 * 1024**3, remaining=7 * 1024**3
|
||
)
|
||
|
||
|
||
def test_drive_id_call_keeps_returning_plain_string() -> None:
|
||
response = _drives_response({"quota": {"used": 1, "total": 2}})
|
||
with patch("app.opencloud_access.requests.get", return_value=response):
|
||
assert resolve_personal_drive_id("raykovmse", "token") == "s$personal"
|
||
|
||
|
||
def test_missing_quota_field_gives_none() -> None:
|
||
with patch("app.opencloud_access.requests.get", return_value=_drives_response({})):
|
||
drive = resolve_personal_drive("raykovmse", "token")
|
||
|
||
assert drive.id == "s$personal"
|
||
assert drive.quota is None
|
||
|
||
|
||
def test_quota_without_used_gives_none() -> None:
|
||
response = _drives_response({"quota": {"total": 100, "state": "normal"}})
|
||
with patch("app.opencloud_access.requests.get", return_value=response):
|
||
assert resolve_personal_drive("raykovmse", "token").quota is None
|
||
|
||
|
||
def test_zero_total_means_no_limit() -> None:
|
||
response = _drives_response({"quota": {"used": 5, "total": 0}})
|
||
with patch("app.opencloud_access.requests.get", return_value=response):
|
||
assert resolve_personal_drive("raykovmse", "token").quota == DriveQuota(
|
||
used=5, total=None, remaining=None
|
||
)
|
||
|
||
|
||
def test_format_quota() -> None:
|
||
assert (
|
||
main._format_quota(DriveQuota(used=int(1.25 * 1024**3), total=10 * 1024**3))
|
||
== "Занято 1,2 из 10 ГБ"
|
||
)
|
||
assert (
|
||
main._format_quota(DriveQuota(used=512 * 1024**2, total=2 * 1024**3))
|
||
== "Занято 0,5 из 2 ГБ"
|
||
)
|
||
assert main._format_quota(DriveQuota(used=0, total=None)) == "Занято 0 ГБ"
|
||
assert main._format_quota(None) is None
|
||
|
||
|
||
def test_load_quota_returns_quota(monkeypatch) -> None:
|
||
monkeypatch.setattr(main, "resolve_opencloud_username", lambda: "raykovmse")
|
||
monkeypatch.setattr(main, "resolve_opencloud_access_token", lambda: "token")
|
||
response = _drives_response({"quota": {"used": 1, "total": 2}})
|
||
with patch("app.opencloud_access.requests.get", return_value=response):
|
||
assert main._load_quota() == DriveQuota(used=1, total=2)
|
||
|
||
|
||
def test_load_quota_swallows_failed_request(monkeypatch) -> None:
|
||
monkeypatch.setattr(main, "resolve_opencloud_username", lambda: "raykovmse")
|
||
monkeypatch.setattr(main, "resolve_opencloud_access_token", lambda: "token")
|
||
with patch(
|
||
"app.opencloud_access.requests.get",
|
||
side_effect=requests.ConnectionError("down"),
|
||
):
|
||
assert main._load_quota() is None
|
||
|
||
|
||
def test_load_quota_swallows_unexpected_errors(monkeypatch) -> None:
|
||
monkeypatch.setattr(main, "resolve_opencloud_username", lambda: "raykovmse")
|
||
monkeypatch.setattr(main, "resolve_opencloud_access_token", lambda: "token")
|
||
|
||
def _boom(username, token):
|
||
raise OpenCloudUploadError("HTTP 500")
|
||
|
||
monkeypatch.setattr(main, "resolve_personal_drive", _boom)
|
||
assert main._load_quota() is None
|
||
|
||
|
||
def test_load_quota_without_token_does_not_call_opencloud(monkeypatch) -> None:
|
||
monkeypatch.setattr(main, "resolve_opencloud_username", lambda: "raykovmse")
|
||
monkeypatch.setattr(main, "resolve_opencloud_access_token", lambda: None)
|
||
with patch("app.opencloud_access.requests.get") as get:
|
||
assert main._load_quota() is None
|
||
get.assert_not_called()
|