cvat-ai--cvat/tests/python/sdk/test_auth_store.py

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

import json
import os
from pathlib import Path

import platformdirs
import pytest
from cvat_sdk.core.auth import (
    AuthStore,
    AuthStoreError,
    ProfileEntry,
    get_auth_store_path,
)
from cvat_sdk.core.utils import is_posix


def test_auth_store_path_matches_platformdirs():
    expected = platformdirs.user_config_path("cvat-sdk", "CVAT.ai") / "auth.json"
    assert get_auth_store_path() == expected
    assert isinstance(get_auth_store_path(), Path)


def _store(tmp_path) -> AuthStore:
    return AuthStore(path=tmp_path / "cvat" / "auth.json")


def test_load_returns_empty_doc_when_file_absent(tmp_path):
    doc = _store(tmp_path)._load()
    assert doc == {"version": 1, "profiles": {}}


def test_save_then_load_roundtrips(tmp_path):
    store = _store(tmp_path)
    store._save({"version": 1, "profiles": {"a": {"server": "https://x", "token": "t"}}})
    assert store._load()["profiles"]["a"]["token"] == "t"


@pytest.mark.skipif(not is_posix(), reason="POSIX permission semantics")
def test_save_creates_0600_file_in_0700_dir(tmp_path):
    store = _store(tmp_path)
    store._save({"version": 1, "profiles": {}})
    path = tmp_path / "cvat" / "auth.json"
    assert (path.stat().st_mode & 0o777) == 0o600
    assert (path.parent.stat().st_mode & 0o777) == 0o700


@pytest.mark.skipif(not is_posix(), reason="POSIX permission semantics")
def test_load_refuses_world_readable_file(tmp_path):
    store = _store(tmp_path)
    store._save({"version": 1, "profiles": {}})
    path = tmp_path / "cvat" / "auth.json"
    os.chmod(path, 0o644)
    with pytest.raises(AuthStoreError, match="permission"):
        AuthStore(path=path)._load()


@pytest.mark.skipif(not is_posix(), reason="POSIX permission semantics")
def test_load_allows_file_with_secure_base_permissions_and_special_bits(tmp_path):
    store = _store(tmp_path)
    store._save({"version": 1, "profiles": {}})
    path = tmp_path / "cvat" / "auth.json"
    # 0o1600 is 0o600 plus the POSIX sticky bit; only the base permission bits matter.
    os.chmod(path, 0o1600)
    assert AuthStore(path=path)._load() == {"version": 1, "profiles": {}}


def test_load_rejects_directory_path(tmp_path):
    path = tmp_path / "cvat" / "auth.json"
    path.mkdir(parents=True)
    if is_posix():
        os.chmod(path.parent, 0o700)
        os.chmod(path, 0o700)
    with pytest.raises(AuthStoreError, match="must be a file"):
        _store(tmp_path)._load()


def test_load_rejects_unknown_version(tmp_path):
    path = tmp_path / "cvat" / "auth.json"
    path.parent.mkdir(parents=True)
    path.write_text(json.dumps({"version": 999, "profiles": {}}))
    if is_posix():
        os.chmod(path.parent, 0o700)
        os.chmod(path, 0o600)
    with pytest.raises(AuthStoreError, match="version"):
        _store(tmp_path)._load()


def test_load_rejects_corrupt_json(tmp_path):
    path = tmp_path / "cvat" / "auth.json"
    path.parent.mkdir(parents=True)
    path.write_text("{not json")
    if is_posix():
        os.chmod(path.parent, 0o700)
        os.chmod(path, 0o600)
    with pytest.raises(AuthStoreError):
        _store(tmp_path)._load()


def _entry(server="https://app.cvat.ai", token="tok") -> ProfileEntry:
    return ProfileEntry(server=server, token=token, created_date="2026-01-01T00:00:00+00:00")


def test_put_get_list_remove_profile(tmp_path):
    store = _store(tmp_path)
    assert store.list_profiles() == {}
    store.put_profile("mycvat", _entry())
    assert store.get_profile("mycvat") == _entry()
    assert set(store.list_profiles()) == {"mycvat"}
    store.remove_profile("mycvat")
    assert store.get_profile("mycvat") is None


def test_failed_write_does_not_pollute_cached_doc(tmp_path, monkeypatch):
    store = _store(tmp_path)
    store.put_profile("kept", _entry())

    def fail_save(_doc):
        raise AuthStoreError("save failed")

    monkeypatch.setattr(store, "_save", fail_save)

    with pytest.raises(AuthStoreError, match="save failed"):
        store.put_profile("ghost", _entry(token="ghost"))

    assert set(store.list_profiles()) == {"kept"}
    assert store.get_profile("ghost") is None


def test_auth_store_reuses_loaded_doc(tmp_path, monkeypatch):
    store = _store(tmp_path)
    store._save(
        {
            "version": 1,
            "profiles": {
                "mycvat": {
                    "server": "https://x",
                    "token": "t",
                    "created_date": "2026-01-01T00:00:00+00:00",
                }
            },
        }
    )

    store = _store(tmp_path)
    read_count = 0
    original_read_text = Path.read_text

    def read_text(self, *args, **kwargs):
        nonlocal read_count
        read_count += 1
        return original_read_text(self, *args, **kwargs)

    monkeypatch.setattr(Path, "read_text", read_text)

    assert store.get_profile("mycvat") is not None
    assert set(store.list_profiles()) == {"mycvat"}
    assert read_count == 1


def test_first_profile_becomes_default_when_requested(tmp_path):
    store = _store(tmp_path)
    store.put_profile("mycvat", _entry(), set_default=True)
    name, entry = store.get_default_profile()
    assert name == "mycvat"
    assert entry == _entry()


def test_first_profile_becomes_default_even_without_flag(tmp_path):
    store = _store(tmp_path)
    store.put_profile("mycvat", _entry())
    assert store.get_default_profile()[0] == "mycvat"


def test_put_profile_after_clear_default_does_not_recreate_default(tmp_path):
    store = _store(tmp_path)
    store.put_profile("first", _entry())

    store.clear_default_profile()
    store.put_profile("second", _entry(token="second"))

    assert store.get_default_profile() is None


def test_set_default_profile_requires_existing(tmp_path):
    store = _store(tmp_path)
    with pytest.raises(KeyError):
        store.set_default_profile("nope")


def test_removing_default_profile_clears_default(tmp_path):
    store = _store(tmp_path)
    store.put_profile("mycvat", _entry(), set_default=True)
    store.remove_profile("mycvat")
    assert store.get_default_profile() is None


def test_put_profile_after_removing_default_with_profiles_remaining_does_not_recreate_default(
    tmp_path,
):
    store = _store(tmp_path)
    store.put_profile("first", _entry())
    store.put_profile("second", _entry(token="second"))

    store.remove_profile("first")
    store.put_profile("third", _entry(token="third"))

    assert store.get_default_profile() is None


def test_default_server_set_get_clear(tmp_path):
    store = _store(tmp_path)
    assert store.get_default_server() is None
    store.set_default_server("https://staging.example.com")
    assert store.get_default_server() == "https://staging.example.com"
    store.clear_default_server()
    assert store.get_default_server() is None


def test_remove_unknown_profile_raises(tmp_path):
    with pytest.raises(KeyError):
        _store(tmp_path).remove_profile("ghost")