Tests: improve test coverage (#175)

* tests: Update manage documents test.

* tests: Update manage spaces tests.

* tests: Added tests for manage organisations module.

* tests: Add tests for manage users.

source/docq/manage_spaces.py

@@ -186,7 +186,7 @@ def get_shared_spaces(space_ids: List[int]) -> List[Tuple[int, int, str, str, bo
         sqlite3.connect(get_sqlite_system_file(), detect_types=sqlite3.PARSE_DECLTYPES)
     ) as connection, closing(connection.cursor()) as cursor:
         placeholders = ", ".join("?" * len(space_ids))
-        query = "SELECT id, org_id, name, summary, archived, datasource_type, datasource_configs, created_at, updated_at FROM spaces WHERE id IN ({})".format(
+        query = "SELECT id, org_id, name, summary, archived, datasource_type, datasource_configs, created_at, updated_at FROM spaces WHERE id IN ({})".format(  # noqa: S608
             placeholders
         )
         cursor.execute(query, space_ids)

tests/unit/docq/manage_documents_test.py

@@ -1,27 +1,16 @@
 """Test manage_documents.py."""
+import os
+import tempfile
 import unittest
 from typing import Self
 from unittest.mock import MagicMock, Mock, patch
 
-# from docq.manage_documents import (
-#     _classify_file_sources,
-#     _classify_web_sources,
-#     _generate_file_markdown,
-#     _generate_web_markdown,
-#     _get_download_link,
-#     _parse_metadata,
-#     _remove_ascii_control_characters,
-#     format_document_sources,
-# )
-
 
 class TestManageDocuments(unittest.TestCase):
     """Test manage_documents."""
 
-    #@patch("docq.support.metadata_extractors.DEFAULT_MODEL_PATH")
     def setUp(self: Self) -> None:
         """Prepare test data."""
-        #mock_DEFAULT_MODEL_PATH.return_value = "./sfdsf"
         self.web_metadata = {
             "source_website": "https://example.net",
             "page_title": "page_title",
@@ -46,6 +35,83 @@ def setUp(self: Self) -> None:
         self.file_source_node.append(Mock(node=file_node, score=1))
         self.source_template = "\n##### Source:\n{file_sources}"
 
+    @patch("docq.manage_documents.reindex")
+    @patch("docq.manage_documents.get_upload_file")
+    def test_upload(self: Self, get_upload_file: Mock, reindex: Mock) -> None:
+        """Test upload."""
+        with tempfile.NamedTemporaryFile() as temp_file:
+            from docq.manage_documents import upload
+
+            get_upload_file.return_value = temp_file.name
+            space = Mock()
+            file_content = bytes("test", "utf-8")
+            upload(temp_file.name, file_content, space)
+
+            reindex.assert_called_once_with(space)
+            get_upload_file.assert_called_once_with(space, temp_file.name)
+            assert os.path.exists(temp_file.name), f"Path {temp_file.name} should exist"
+            assert os.path.isfile(temp_file.name), f"File {temp_file.name} should be a file"
+            assert os.path.getsize(temp_file.name) == len(file_content), f"File {temp_file.name} should have content"
+
+    @patch("docq.manage_documents.get_upload_file")
+    def test_get_file(self: Self, get_upload_file: Mock) -> None:
+        """Test get_file."""
+        from docq.manage_documents import get_file
+
+        space = Mock()
+        file_name = "file_name"
+        get_upload_file.return_value = file_name
+        assert get_file(file_name, space) == file_name, "File name should match"
+        get_upload_file.assert_called_once_with(space, file_name)
+
+    @patch("docq.manage_documents.get_upload_file")
+    def test_delete(self: Self, get_upload_file: Mock) -> None:
+        """Test delete."""
+        from docq.manage_documents import delete
+
+        space = Mock()
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_name = os.path.join(temp_dir, "file_name")
+            ctrl_file_name = os.path.join(temp_dir, ".file_name")
+            open(file_name, "w").close()
+            open(ctrl_file_name, "w").close()
+
+            assert os.path.exists(file_name), f"File {file_name} should exist"
+            get_upload_file.return_value = file_name
+            delete(file_name, space)
+
+            assert not os.path.exists(file_name), f"File {file_name} should not exist"
+            assert os.path.exists(ctrl_file_name), f"Control file {ctrl_file_name} should exist"
+            get_upload_file.assert_called_once_with(space, file_name)
+
+    @patch("docq.manage_documents.reindex")
+    @patch("docq.manage_documents.get_upload_dir")
+    def test_delete_all(self: Self, get_upload_dir: Mock, reindex: Mock) -> None:
+        """Test delete_all."""
+        from docq.manage_documents import delete_all
+
+        space = Mock()
+        with tempfile.TemporaryDirectory() as temp_dir:
+            upload_dir = os.path.join(temp_dir, "upload")
+            os.mkdir(upload_dir)
+
+            assert os.path.exists(upload_dir), f"Directory path {upload_dir} should exist"
+            assert os.path.isdir(upload_dir), f"Directory {upload_dir} should be a directory"
+            get_upload_dir.return_value = temp_dir
+            delete_all(space)
+
+            assert not os.path.exists(upload_dir), f"Directory {temp_dir} should not exist"
+            get_upload_dir.assert_called_once_with(space)
+            reindex.assert_called_once_with(space)
+
+    def test_is_web_address(self: Self) -> None:
+        """Test _is_web_address."""
+        from docq.manage_documents import _is_web_address
+
+        assert _is_web_address("http://example.net"), "http://example.net should pass as web address"
+        assert _is_web_address("https://example.net"), "https://example.net should pass as web address"
+        assert not _is_web_address("example.net"), "example.net should not pass as web address"
+
     def test_remove_ascii_control_characters(self: Self) -> None:
         """Test _remove_ascii_control_characters."""
         from docq.manage_documents import _remove_ascii_control_characters
@@ -86,6 +152,8 @@ def test_classify_web_sources(self: Self) -> None:
 
         sources = _classify_web_sources("website", "uri", "page_title")
         assert sources == {"website": [("page_title", "uri")]}
+        sources = _classify_web_sources("website", "uri", "page_title", sources)
+        assert sources == {"website": [("page_title", "uri"), ("page_title", "uri")]}
 
     @patch("docq.manage_documents._get_download_link")
     def test_generate_file_markdown(self: Self, download_link: Mock) -> None:
@@ -104,11 +172,27 @@ def test_generate_web_markdown(self: Self) -> None:
         sources = {"website": [("page_title", "uri"), ("page_title", "uri")]}
         assert _generate_web_markdown(sources) == "\n> ###### website\n>- [page_title](uri)\n\n"
 
-    def test_get_download_link(self: Self) -> None:
+    @patch("docq.manage_documents.runtime")
+    def test_get_download_link(self: Self, runtime: Mock) -> None:
         """Test _get_download_link."""
         from docq.manage_documents import _get_download_link
 
-        assert _get_download_link("name", "uri") == "#"
+        web_address = "https://example.net"
+
+        assert _get_download_link("name", web_address) == web_address, "Web address should return the same address"
+
+        runtime.exists = MagicMock(return_value=False)
+        assert _get_download_link("name", "uri") == "#", "Download link should return '#' if runtime does not exist"
+
+        add = MagicMock(return_value="https://some_link.host")
+        get_instance = MagicMock(
+            return_value=Mock(media_file_mgr=Mock(add=add))
+        )
+
+        runtime.exists = MagicMock(return_value=True)
+        runtime.get_instance = get_instance
+        with tempfile.NamedTemporaryFile() as temp_file:
+            assert _get_download_link("name", temp_file.name) == "https://some_link.host", "Download link should be returned if runtime exists."
 
     @patch("docq.manage_documents._get_download_link")
     def test_format_document_sources(self: Self, file_link: Mock) -> None:
@@ -124,4 +208,7 @@ def test_format_document_sources(self: Self, file_link: Mock) -> None:
         assert format_document_sources(self.web_source_node) == self.source_template.format(
             file_sources=web_template
         ), "Web source should return a matching web template"
-        assert format_document_sources({}) == "", "Empty source should return empty string"
+        assert format_document_sources([]) == "", "Empty source should return empty string"
+
+        with patch("docq.manage_documents._parse_metadata", MagicMock(return_value=("name", "page", "uri", "UNKNOWN_DS"))):
+            assert format_document_sources(self.file_source_node) == "", "Unknown data source should return empty string"

tests/unit/docq/manage_organisations_test.py

@@ -0,0 +1,196 @@
+"""Test for docq.manage_organisations module."""
+import logging as log
+import os
+import sqlite3
+import tempfile
+from contextlib import closing, suppress
+from typing import Generator, Optional
+from unittest.mock import patch
+
+import pytest
+from docq.constants import DEFAULT_ORG_ID
+
+TEST_USER_ID = 1005
+
+@pytest.fixture(scope="session")
+def manage_orgs_test_dir() -> Generator:
+    """Return the sqlite system file."""
+    from docq.manage_organisations import _init
+
+    log.info("Setup manage organisations tests...")
+
+    with tempfile.TemporaryDirectory() as temp_dir, patch("docq.manage_organisations.get_sqlite_system_file"
+        ) as get_sqlite_system_file:
+        sqlite_system_file = os.path.join(temp_dir, "system.db")
+        get_sqlite_system_file.return_value = sqlite_system_file
+        _init()
+
+        yield  temp_dir, sqlite_system_file, get_sqlite_system_file,
+
+    log.info("Teardown manage organisations tests...")
+
+
+def insert_test_org(sqlite_system_file: str, name: str) -> Optional[int]:
+    """Insert test org."""
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        cursor.execute(
+            "INSERT INTO orgs (name) VALUES (?)",
+            (name,),
+        )
+        org_id = cursor.lastrowid
+        connection.commit()
+        return org_id
+
+
+def test_db_init(manage_orgs_test_dir: tuple) -> None:
+    """Test database init."""
+    sqlite_system_file = manage_orgs_test_dir[1]
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        sql_select = "SELECT name FROM sqlite_master WHERE type='table' AND name = ?"
+        cursor.execute(sql_select, ("orgs",))
+
+        assert cursor.fetchone() is not None, "Table orgs should exist"
+
+
+@pytest.mark.first()
+def test_init_default_org_if_necessary() -> None:
+    """Test init_default_org_if_necessary."""
+    from docq.manage_organisations import _init_default_org_if_necessary
+
+    with patch("docq.manage_organisations.manage_settings._init_default_system_settings"
+        ) as _init_default_system_settings, patch("docq.manage_organisations.manage_settings._init_default_org_settings"
+        ) as _init_default_org_settings:
+        assert _init_default_org_if_necessary() is True, "Default org should be created"
+
+        _init_default_system_settings.assert_called_once()
+        _init_default_org_settings.assert_called_once_with(DEFAULT_ORG_ID)
+
+        assert _init_default_org_if_necessary() is False, "Default org should not be created"
+
+
+def test_list_organisations(manage_orgs_test_dir: tuple) -> None:
+    """Test list_organisations."""
+    from docq.manage_organisations import list_organisations
+    from docq.manage_users import _init
+
+    sqlite_system_file = manage_orgs_test_dir[1]
+    with patch("docq.manage_users.get_sqlite_system_file") as get_sqlite_system_file:
+        get_sqlite_system_file.return_value = sqlite_system_file
+        _init()
+
+    org_names = ["Test_list_organisations_org_1", "Test_list_organisations_org_2"]
+    org_id = insert_test_org(sqlite_system_file, org_names[0])
+
+    assert org_id is not None, "The sample list organisations test org should not be None"
+    org_list = list_organisations(name_match=org_names[0])
+
+    assert org_list is not None, "The sample list organisations test org should not be None"
+    assert len(org_list) == 1, "Expected only one org to be returned."
+    assert org_list[0][0] == org_id, "The sample list organisations test org id should match"
+
+    org_id = insert_test_org(sqlite_system_file, org_names[1])
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        cursor.execute("INSERT INTO users (username, password, fullname) VALUES (?, ?, ?)", ("list_orgs_test_user", "test_password", "Test User"))
+        user_id = cursor.lastrowid
+
+        assert user_id is not None, "The sample list organisations test user should not be None"
+        cursor.execute(
+            "INSERT INTO org_members (user_id, org_id) VALUES (?, ?)",
+            (user_id, org_id),
+        )
+        connection.commit()
+
+    org_list = list_organisations(user_id=user_id)
+    assert org_list is not None, "The sample list organisations test org should not be None"
+    assert len(org_list) == 1, "Expected only one org to be returned."
+    assert org_list[0][0] == org_id, "The sample list organisations test org id should match"
+
+
+def test_create_organisation_sql(manage_orgs_test_dir: tuple) -> None:
+    """Test _create_organisation_sql."""
+    from docq.manage_organisations import _create_organisation_sql
+
+    org_name = "Test_create_organisation_sql_org"
+    sqlite_system_file = manage_orgs_test_dir[1]
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        cursor = _create_organisation_sql(cursor, org_name)
+        connection.commit()
+
+        cursor.execute("SELECT name FROM orgs WHERE name = ?", (org_name,))
+        assert cursor.fetchone() is not None, "The sample create organisation test org should exist"
+
+
+def test_create_organisation(manage_orgs_test_dir: tuple) -> None:
+    """Test create_organisation."""
+    from docq.manage_organisations import create_organisation
+
+    org_name = "Test_create_organisation_org"
+    sqlite_system_file = manage_orgs_test_dir[1]
+
+    with patch("docq.manage_organisations.manage_users._add_organisation_member_sql"
+        ) as _add_organisation_member_sql, patch("docq.manage_organisations.manage_settings._init_default_org_settings"
+        ) as _init_default_org_settings:
+        org_id = create_organisation(org_name, TEST_USER_ID)
+
+        assert org_id is not None, "The sample create organisation test org should not be None"
+        _add_organisation_member_sql.assert_called_once()
+        _init_default_org_settings.assert_called_once_with(org_id)
+
+        with suppress(Exception):
+            duplicate = create_organisation(org_name, TEST_USER_ID)
+            assert duplicate is None, "Creating org with duplicate name should fail."
+
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        cursor.execute("SELECT name FROM orgs WHERE name = ?", (org_name,))
+        assert cursor.fetchone() is not None, "The sample create organisation test org should exist"
+
+
+def test_update_organisation(manage_orgs_test_dir: tuple) -> None:
+    """Test update_organisation."""
+    from docq.manage_organisations import update_organisation
+
+    org_name = "Test_update_organisation_org"
+    org_name_updated = "Test_update_organisation_org_updated"
+    sqlite_system_file = manage_orgs_test_dir[1]
+    org_id = insert_test_org(sqlite_system_file, org_name)
+    ctrl_org_name = "Test_update_organisation_ctrl_org"
+    ctrl_org_id = insert_test_org(sqlite_system_file, ctrl_org_name)
+
+    assert org_id is not None, "The sample update organisation test org should not be None"
+    assert ctrl_org_id is not None, "The sample update organisation control org should not be None"
+    assert update_organisation(org_id, org_name_updated ) is True, "The sample update organisation test org should be updated"
+
+    with suppress(Exception):
+        assert update_organisation(org_id, ctrl_org_name) is False, "Updating org with duplicate name should fail."
+
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        cursor.execute("SELECT name FROM orgs WHERE name = ?", (org_name_updated,))
+        assert cursor.fetchone() is not None, "The updated org should exist"
+
+        cursor.execute("SELECT id, name FROM orgs WHERE name = ?", (ctrl_org_name,))
+        org = cursor.fetchone()
+        assert org is not None, "The control org should exist"
+        assert org[0] == ctrl_org_id, "The control org id should match"
+
+
+def test_archive_organisation(manage_orgs_test_dir: tuple) -> None:
+    """Test archive_organisation."""
+    from docq.manage_organisations import archive_organisation
+
+    org_name = "Test_archive_organisation_org"
+    sqlite_system_file = manage_orgs_test_dir[1]
+    org_id = insert_test_org(sqlite_system_file, org_name)
+
+    assert org_id is not None, "The sample archive organisation test org should not be None"
+    assert archive_organisation(org_id) is True, "The sample archive organisation test org should be archived"
+
+    with closing(sqlite3.connect(sqlite_system_file, detect_types=sqlite3.PARSE_DECLTYPES)) as connection, closing(connection.cursor()) as cursor:
+        cursor.execute("SELECT name, archived FROM orgs WHERE name = ?", (org_name,))
+        org = cursor.fetchone()
+
+    assert org is not None, "The archived org should not exist"
+    assert org[0] == org_name, "The archived org name should match"
+    assert org[1] == 1, "The archived org should be archived"
+
+
+