Add integration tests

Author: uroni

.github/workflows/integration-tests.yml

@@ -0,0 +1,37 @@
+name: Integration Tests
+
+on:
+  push:
+    branches: [ main, master ]
+  pull_request:
+    branches: [ main, master ]
+
+jobs:
+  integration-tests:
+    runs-on: ubuntu-24.04
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.x'
+
+      - name: Install urbackup-server
+        run: |
+          echo 'deb http://download.opensuse.org/repositories/home:/uroni/xUbuntu_24.04/ /' | sudo tee /etc/apt/sources.list.d/home:uroni.list
+          curl -fsSL https://download.opensuse.org/repositories/home:uroni/xUbuntu_24.04/Release.key | gpg --dearmor | sudo tee /etc/apt/trusted.gpg.d/home_uroni.gpg > /dev/null
+          sudo apt-get update
+          sudo apt-get install -y urbackup-server
+
+      - name: Start urbackup-server
+        run: sudo systemctl start urbackupsrv
+
+      - name: Install Python dependencies
+        run: |
+          pip install pytest
+          pip install -e .
+
+      - name: Run integration tests
+        run: pytest

.gitignore

@@ -4,3 +4,4 @@ urbackup_api/__pycache__/*
 
 # IDE
 .idea
+tests/__pycache__

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+testpaths = tests

tests/conftest.py

@@ -0,0 +1,52 @@
+import subprocess
+import time
+
+import pytest
+
+import urbackup_api
+
+
+SERVER_URL = "http://127.0.0.1:55414/x"
+ADMIN_USER = "admin"
+ADMIN_PASSWORD = "test1234"
+
+
+def _restart_clean_server():
+    """Stop urbackupsrv, wipe /var/urbackup/, start fresh."""
+    subprocess.run(["sudo", "systemctl", "stop", "urbackupsrv"], check=True)
+    subprocess.run(["sudo", "rm", "-rf", "/var/urbackup/"], check=True)
+    subprocess.run(["sudo", "mkdir", "-p", "/var/urbackup/"], check=True)
+    subprocess.run(["sudo", "chown", "urbackup:urbackup", "/var/urbackup/"], check=True)
+    subprocess.run(["sudo", "systemctl", "start", "urbackupsrv"], check=True)
+    # Wait for the server to be ready
+    for _ in range(30):
+        try:
+            s = urbackup_api.urbackup_server(SERVER_URL, ADMIN_USER, "")
+            if s.login():
+                break
+        except Exception:
+            pass
+        time.sleep(1)
+    else:
+        raise RuntimeError("urbackupsrv did not become ready after restart")
+    # Set admin password
+    subprocess.run(
+        ["sudo", "urbackupsrv", "reset-admin-pw", "-p", ADMIN_PASSWORD],
+        check=True,
+    )
+
+
+@pytest.fixture(scope="module", autouse=True)
+def clean_server():
+    """Restart urbackupsrv with clean data before each test module."""
+    _restart_clean_server()
+    yield
+    # No teardown needed; next module will restart anyway
+
+
+@pytest.fixture()
+def server():
+    """Return a logged-in urbackup_server instance."""
+    s = urbackup_api.urbackup_server(SERVER_URL, ADMIN_USER, ADMIN_PASSWORD)
+    assert s.login(), "Failed to login to urbackup server"
+    return s

tests/test_backups.py

@@ -0,0 +1,85 @@
+"""Tests for backup-related operations (backups listing, actions, start/stop)."""
+
+import pytest
+
+
+class TestBackupLists:
+
+    def test_get_clientbackups_empty(self, server):
+        server.add_client("pytest-backup-client")
+        status = server.get_client_status("pytest-backup-client")
+        backups = server.get_clientbackups(status["id"])
+        assert isinstance(backups, list)
+        assert backups == []
+
+    def test_get_clientimagebackups_empty(self, server):
+        server.add_client("pytest-imgbackup-client")
+        status = server.get_client_status("pytest-imgbackup-client")
+        backups = server.get_clientimagebackups(status["id"])
+        assert isinstance(backups, list)
+        assert backups == []
+
+
+class TestStartBackup:
+
+    @pytest.fixture(autouse=True)
+    def _create_client(self, server):
+        server.add_client("pytest-startbackup-client")
+
+    def test_start_incr_file_backup_no_online_client(self, server):
+        # Client exists but is not online, so backup start should fail
+        result = server.start_incr_file_backup("pytest-startbackup-client")
+        assert result is False
+
+    def test_start_full_file_backup_no_online_client(self, server):
+        result = server.start_full_file_backup("pytest-startbackup-client")
+        assert result is False
+
+    def test_start_incr_image_backup_no_online_client(self, server):
+        result = server.start_incr_image_backup("pytest-startbackup-client")
+        assert result is False
+
+    def test_start_full_image_backup_no_online_client(self, server):
+        result = server.start_full_image_backup("pytest-startbackup-client")
+        assert result is False
+
+    def test_start_backup_nonexistent_client(self, server):
+        result = server.start_incr_file_backup("nonexistent-client-xyz")
+        assert result is False
+
+
+class TestActions:
+
+    def test_get_actions_empty(self, server):
+        actions = server.get_actions()
+        assert isinstance(actions, list)
+
+    def test_stop_action_missing_fields(self, server):
+        # action dict missing required keys
+        result = server.stop_action({})
+        assert result is False
+
+        result = server.stop_action({"clientid": 1})
+        assert result is False
+
+        result = server.stop_action({"id": 1})
+        assert result is False
+
+
+class TestActionConstants:
+
+    def test_action_constants_exist(self):
+        from urbackup_api import urbackup_server
+
+        assert urbackup_server.action_incr_file == 1
+        assert urbackup_server.action_full_file == 2
+        assert urbackup_server.action_incr_image == 3
+        assert urbackup_server.action_full_image == 4
+        assert urbackup_server.action_resumed_incr_file == 5
+        assert urbackup_server.action_resumed_full_file == 6
+        assert urbackup_server.action_file_restore == 8
+        assert urbackup_server.action_image_restore == 9
+        assert urbackup_server.action_client_update == 10
+        assert urbackup_server.action_check_db_integrity == 11
+        assert urbackup_server.action_backup_db == 12
+        assert urbackup_server.action_recalc_stats == 13

tests/test_clients.py

@@ -0,0 +1,87 @@
+"""Tests for client management (add, status, settings, authkey)."""
+
+import pytest
+
+
+class TestAddClient:
+
+    def test_add_client(self, server):
+        ret = server.add_client("pytest-client-1")
+        assert ret is not None
+        assert ret["added_new_client"] is True
+        assert "new_authkey" in ret
+        assert ret["new_clientname"] == "pytest-client-1"
+
+    def test_add_client_duplicate_returns_none(self, server):
+        server.add_client("pytest-dup")
+        ret = server.add_client("pytest-dup")
+        assert ret is None
+
+    def test_add_client_with_group(self, server):
+        ret = server.add_client("pytest-grouped", groupname="testgroup")
+        assert ret is not None
+        assert ret["added_new_client"] is True
+
+
+class TestGetClientStatus:
+
+    @pytest.fixture(autouse=True)
+    def _create_client(self, server):
+        server.add_client("pytest-status-client")
+
+    def test_get_client_status(self, server):
+        status = server.get_client_status("pytest-status-client")
+        assert status is not None
+        assert status["name"] == "pytest-status-client"
+        assert "id" in status
+
+    def test_get_client_status_nonexistent(self, server):
+        status = server.get_client_status("nonexistent-client-xyz")
+        assert status is None
+
+    def test_client_appears_in_status_list(self, server):
+        all_status = server.get_status()
+        names = [c["name"] for c in all_status]
+        assert "pytest-status-client" in names
+
+
+class TestClientSettings:
+
+    @pytest.fixture(autouse=True)
+    def _create_client(self, server):
+        server.add_client("pytest-settings-client")
+
+    def test_get_client_settings(self, server):
+        settings = server.get_client_settings("pytest-settings-client")
+        assert settings is not None
+        assert isinstance(settings, dict)
+        assert "internet_authkey" in settings
+
+    def test_get_client_settings_nonexistent(self, server):
+        settings = server.get_client_settings("nonexistent-client-xyz")
+        assert settings is None
+
+    def test_change_client_setting(self, server):
+        result = server.change_client_setting(
+            "pytest-settings-client", "internet_authkey", "my-custom-key"
+        )
+        assert result is True
+
+        authkey = server.get_client_authkey("pytest-settings-client")
+        assert authkey is not None
+        assert authkey["value"] == "my-custom-key"
+
+
+class TestClientAuthkey:
+
+    @pytest.fixture(autouse=True)
+    def _create_client(self, server):
+        server.add_client("pytest-authkey-client")
+
+    def test_get_client_authkey(self, server):
+        authkey = server.get_client_authkey("pytest-authkey-client")
+        assert authkey is not None
+
+    def test_get_client_authkey_nonexistent(self, server):
+        authkey = server.get_client_authkey("nonexistent-client-xyz")
+        assert authkey is None

tests/test_extra_clients.py

@@ -0,0 +1,39 @@
+"""Tests for extra clients management."""
+
+
+class TestExtraClients:
+
+    def test_get_extra_clients_empty(self, server):
+        extras = server.get_extra_clients()
+        assert isinstance(extras, list)
+        assert extras == []
+
+    def test_add_extra_client(self, server):
+        result = server.add_extra_client("10.0.0.100")
+        assert result is True
+
+        extras = server.get_extra_clients()
+        hostnames = [e["hostname"] for e in extras]
+        assert "10.0.0.100" in hostnames
+
+    def test_add_multiple_extra_clients(self, server):
+        server.add_extra_client("10.0.0.101")
+        server.add_extra_client("10.0.0.102")
+
+        extras = server.get_extra_clients()
+        hostnames = [e["hostname"] for e in extras]
+        assert "10.0.0.101" in hostnames
+        assert "10.0.0.102" in hostnames
+
+    def test_remove_extra_client(self, server):
+        server.add_extra_client("10.0.0.200")
+        extras = server.get_extra_clients()
+        target = [e for e in extras if e["hostname"] == "10.0.0.200"]
+        assert len(target) == 1
+
+        result = server.remove_extra_client(target[0]["id"])
+        assert result is True
+
+        extras_after = server.get_extra_clients()
+        hostnames_after = [e["hostname"] for e in extras_after]
+        assert "10.0.0.200" not in hostnames_after

tests/test_installer.py

@@ -0,0 +1,42 @@
+"""Tests for installer download functionality."""
+
+import os
+import tempfile
+
+from urbackup_api import installer_os
+
+
+class TestDownloadInstaller:
+
+    def test_download_linux_installer(self, server):
+        with tempfile.NamedTemporaryFile(suffix=".sh", delete=False) as tmp:
+            tmpname = tmp.name
+
+        try:
+            result = server.download_installer(tmpname, "pytest-installer-client", installer_os.Linux)
+            assert result is True
+            assert os.path.exists(tmpname)
+            assert os.path.getsize(tmpname) > 0
+        finally:
+            os.unlink(tmpname)
+
+    def test_download_installer_existing_client(self, server):
+        # First call creates the client
+        server.add_client("pytest-installer-existing")
+
+        with tempfile.NamedTemporaryFile(suffix=".sh", delete=False) as tmp:
+            tmpname = tmp.name
+
+        try:
+            result = server.download_installer(tmpname, "pytest-installer-existing", installer_os.Linux)
+            assert result is True
+            assert os.path.getsize(tmpname) > 0
+        finally:
+            os.unlink(tmpname)
+
+
+class TestInstallerOsEnum:
+
+    def test_installer_os_values(self):
+        assert installer_os.Windows.value == ("windows",)
+        assert installer_os.Linux.value == "linux"

tests/test_livelog.py

@@ -0,0 +1,18 @@
+"""Tests for live log retrieval."""
+
+
+class TestLiveLog:
+
+    def test_get_livelog(self, server):
+        log = server.get_livelog()
+        # Fresh server should have some log entries (startup messages)
+        # or None if no entries exist yet
+        if log is not None:
+            assert isinstance(log, list)
+            for entry in log:
+                assert "id" in entry
+
+    def test_get_livelog_updates_lastlogid(self, server):
+        log = server.get_livelog()
+        if log is not None:
+            assert server._lastlogid > 0