better auth tests, refactoring

3 files changed, 242 insertions, 0 deletions

diff --git a/tests/apps/__init__.py b/tests/apps/__init__.py
new file mode 100644
index 00000000..1e91f7fa
--- /dev/null
+++ b/tests/apps/__init__.py
@@ -0,0 +1,20 @@
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main Pi-KVM daemon.                                          #
+#                                                                            #
+#    Copyright (C) 2018  Maxim Devaev <[email protected]>                    #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+# ========================================================================== #
diff --git a/tests/apps/test_cleanup.py b/tests/apps/test_cleanup.py
new file mode 100644
index 00000000..3e2e4c72
--- /dev/null
+++ b/tests/apps/test_cleanup.py
@@ -0,0 +1,71 @@
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main Pi-KVM daemon.                                          #
+#                                                                            #
+#    Copyright (C) 2018  Maxim Devaev <[email protected]>                    #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+# ========================================================================== #
+
+
+import os
+import secrets
+import multiprocessing
+import multiprocessing.queues
+import time
+
+import setproctitle
+
+from kvmd.apps.cleanup import main
+
+
+# =====
+def test_ok(tmpdir) -> None:  # type: ignore
+    queue: multiprocessing.queues.Queue = multiprocessing.Queue()
+
+    ustreamer_tmp_path = os.path.abspath(str(tmpdir.join("ustr-" + secrets.token_hex(3))))
+    os.symlink("/usr/bin/ustreamer", ustreamer_tmp_path)
+
+    ustreamer_sock_path = os.path.abspath(str(tmpdir.join("ustreamer-fake.sock")))
+    open(ustreamer_sock_path, "w").close()
+    kvmd_sock_path = os.path.abspath(str(tmpdir.join("kvmd-fake.sock")))
+    open(kvmd_sock_path, "w").close()
+
+    def ustreamer_fake() -> None:
+        setproctitle.setproctitle(os.path.basename(ustreamer_tmp_path))
+        queue.put(True)
+        while True:
+            time.sleep(1)
+
+    proc = multiprocessing.Process(target=ustreamer_fake, daemon=True)
+    proc.start()
+    assert queue.get(timeout=5)
+
+    assert proc.is_alive()
+    main([
+        "kvmd-cleanup",
+        "--set-options",
+        "kvmd/server/port=0",
+        "kvmd/server/unix=" + kvmd_sock_path,
+        "kvmd/streamer/port=0",
+        "kvmd/streamer/unix=" + ustreamer_sock_path,
+        "kvmd/streamer/cmd=" + ustreamer_tmp_path,
+    ])
+
+    assert not os.path.exists(ustreamer_sock_path)
+    assert not os.path.exists(kvmd_sock_path)
+
+    assert not proc.is_alive()
+    proc.join()
diff --git a/tests/apps/test_htpasswd.py b/tests/apps/test_htpasswd.py
new file mode 100644
index 00000000..11895c85
--- /dev/null
+++ b/tests/apps/test_htpasswd.py
@@ -0,0 +1,151 @@
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main Pi-KVM daemon.                                          #
+#                                                                            #
+#    Copyright (C) 2018  Maxim Devaev <[email protected]>                    #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+# ========================================================================== #
+
+
+import os
+import hashlib
+import tempfile
+import builtins
+import getpass
+
+from typing import List
+from typing import Generator
+from typing import Any
+
+import passlib.apache
+
+import pytest
+
+from kvmd.apps.htpasswd import main
+
+
+# =====
+def _make_passwd(user: str) -> str:
+    return hashlib.md5(user.encode()).hexdigest()
+
+
[email protected](name="htpasswd", params=[[], ["admin"], ["admin", "user"]])
+def _htpasswd_fixture(request) -> Generator[passlib.apache.HtpasswdFile, None, None]:  # type: ignore
+    (fd, path) = tempfile.mkstemp()
+    os.close(fd)
+    htpasswd = passlib.apache.HtpasswdFile(path)
+    for user in request.param:
+        htpasswd.set_password(user, _make_passwd(user))
+    htpasswd.save()
+    yield htpasswd
+    os.remove(path)
+
+
+def _run_htpasswd(htpasswd: passlib.apache.HtpasswdFile, cmd: List[str]) -> None:
+    main([
+        "kvmd-htpasswd",
+        *cmd,
+        "--set-options",
+        "kvmd/auth/internal/file=" + htpasswd.path,
+    ])
+
+
+# =====
+def test_ok__list(htpasswd: passlib.apache.HtpasswdFile, capsys) -> None:  # type: ignore
+    _run_htpasswd(htpasswd, ["list"])
+    (out, err) = capsys.readouterr()
+    assert len(err) == 0
+    assert sorted(filter(None, out.split("\n"))) == sorted(htpasswd.users()) == sorted(set(htpasswd.users()))
+
+
+# =====
+def test_ok__set_change_stdin(htpasswd: passlib.apache.HtpasswdFile, mocker) -> None:  # type: ignore
+    old_users = set(htpasswd.users())
+    if old_users:
+        assert htpasswd.check_password("admin", _make_passwd("admin"))
+
+        mocker.patch.object(builtins, "input", (lambda: " test "))
+        _run_htpasswd(htpasswd, ["set", "admin", "--read-stdin"])
+
+        htpasswd.load(force=True)
+        assert htpasswd.check_password("admin", " test ")
+        assert old_users == set(htpasswd.users())
+
+
+def test_ok__set_add_stdin(htpasswd: passlib.apache.HtpasswdFile, mocker) -> None:  # type: ignore
+    old_users = set(htpasswd.users())
+    if old_users:
+        mocker.patch.object(builtins, "input", (lambda: " test "))
+        _run_htpasswd(htpasswd, ["set", "new", "--read-stdin"])
+
+        htpasswd.load(force=True)
+        assert htpasswd.check_password("new", " test ")
+        assert old_users.union(["new"]) == set(htpasswd.users())
+
+
+# =====
+def test_ok__set_change_getpass(htpasswd: passlib.apache.HtpasswdFile, mocker) -> None:  # type: ignore
+    old_users = set(htpasswd.users())
+    if old_users:
+        assert htpasswd.check_password("admin", _make_passwd("admin"))
+
+        mocker.patch.object(getpass, "getpass", (lambda *_, **__: " test "))
+        _run_htpasswd(htpasswd, ["set", "admin"])
+
+        htpasswd.load(force=True)
+        assert htpasswd.check_password("admin", " test ")
+        assert old_users == set(htpasswd.users())
+
+
+def test_fail__set_change_getpass(htpasswd: passlib.apache.HtpasswdFile, mocker) -> None:  # type: ignore
+    old_users = set(htpasswd.users())
+    if old_users:
+        assert htpasswd.check_password("admin", _make_passwd("admin"))
+
+        count = 0
+
+        def fake_getpass(*_: Any, **__: Any) -> str:
+            nonlocal count
+            assert count <= 1
+            if count == 0:
+                passwd = " test "
+            else:
+                passwd = "test "
+            count += 1
+            return passwd
+
+        mocker.patch.object(getpass, "getpass", fake_getpass)
+        with pytest.raises(SystemExit, match="Sorry, passwords do not match"):
+            _run_htpasswd(htpasswd, ["set", "admin"])
+        assert count == 2
+
+        htpasswd.load(force=True)
+        assert htpasswd.check_password("admin", _make_passwd("admin"))
+        assert old_users == set(htpasswd.users())
+
+
+# =====
+def test_ok__del(htpasswd: passlib.apache.HtpasswdFile) -> None:
+    old_users = set(htpasswd.users())
+
+    if old_users:
+        assert htpasswd.check_password("admin", _make_passwd("admin"))
+
+    _run_htpasswd(htpasswd, ["del", "admin"])
+
+    htpasswd.load(force=True)
+    assert not htpasswd.check_password("admin", _make_passwd("admin"))
+    assert old_users.difference(["admin"]) == set(htpasswd.users())