Fix settings env preservation, add Forge legacy redirects, and align KoSync safety tests

Author: cporcellijr

src/utils/config_loader.py

@@ -152,8 +152,13 @@ def load_settings(db_service: DatabaseService):
                 # Apply validation or type conversion if needed (mostly string for env vars)
                 val_str = str(value) if value is not None else ""
 
-                # Update environment variable
-                os.environ[key] = val_str
+                # Preserve existing non-empty env vars when DB value is blank.
+                if val_str != "":
+                    os.environ[key] = val_str
+                else:
+                    existing_env = os.environ.get(key, "")
+                    if not existing_env:
+                        os.environ[key] = ""
 
                 # Mask secrets in logs
                 log_val = "******" if any(s in key for s in ['KEY', 'PASSWORD', 'TOKEN']) else val_str

src/web_server.py

@@ -2105,10 +2105,15 @@ def create_app(test_container=None):
     app.context_processor(inject_global_vars)
     app.jinja_env.globals['safe_folder_name'] = safe_folder_name
 
+    def _legacy_book_linker_redirect(dummy=None):
+        return redirect(url_for('forge'), code=301)
+
     # Register all routes here
     app.add_url_rule('/', 'index', index)
     app.add_url_rule('/shelfmark', 'shelfmark', shelfmark)
     app.add_url_rule('/forge', 'forge', forge)
+    app.add_url_rule('/book-linker', 'book_linker_legacy', _legacy_book_linker_redirect)
+    app.add_url_rule('/book-linker/<path:dummy>', 'book_linker_legacy_path', _legacy_book_linker_redirect)
     app.add_url_rule('/match', 'match', match, methods=['GET', 'POST'])
     app.add_url_rule('/batch-match', 'batch_match', batch_match, methods=['GET', 'POST'])
     app.add_url_rule('/delete/<abs_id>', 'delete_mapping', delete_mapping, methods=['POST'])

tests/base_sync_test.py

@@ -190,6 +190,7 @@ def run_sync_test_with_leader_verification(self):
             xpath=f"/html/body/div[1]/p[{int(self.expected_final_pct * 25)}]",
             match_index=int(self.expected_final_pct * 20)
         )
+        self._mock_locator_xpath = mock_locator.xpath
         mocks['ebook_parser'].find_text_location.return_value = mock_locator
         mocks['ebook_parser'].get_perfect_ko_xpath.return_value = mock_locator.xpath
 
@@ -292,7 +293,16 @@ def verify_common_assertions(self, mocks, manager):
                                manager.sync_clients['ABS']._update_abs_progress_with_offset.called)
                 self.assertTrue(abs_updated, "ABS update was not called")
             if leader != 'KOSYNC':
-                self.assertTrue(mocks['kosync_client'].update_progress.called, "KoSync update_progress was not called")
+                # Base tests intentionally use malformed KoSync-style XPath (/html/...),
+                # which should now be skipped by KoSync safety logic.
+                ko_xpath = getattr(mocks['ebook_parser'].find_text_location.return_value, 'xpath', '')
+                if isinstance(ko_xpath, str) and ko_xpath.startswith('/html/'):
+                    self.assertFalse(
+                        mocks['kosync_client'].update_progress.called,
+                        "KoSync update_progress should be skipped for malformed XPath"
+                    )
+                else:
+                    self.assertTrue(mocks['kosync_client'].update_progress.called, "KoSync update_progress was not called")
             if leader != 'STORYTELLER':
                 # [UPDATED] Check update_position
                 if self.test_book.storyteller_uuid:
@@ -363,11 +373,20 @@ def verify_final_state(self, manager):
         expected_pct = self.expected_final_pct
         tolerance = 0.02
 
+        kosync_skipped = (
+            self.get_expected_leader().upper() != "KOSYNC"
+            and isinstance(getattr(self, '_mock_locator_xpath', ''), str)
+            and getattr(self, '_mock_locator_xpath', '').startswith('/html/')
+        )
+
         if self.get_expected_leader() != "None":
             self.assertAlmostEqual(abs_pct, expected_pct, delta=tolerance,
                                    msg=f"ABS final state {abs_pct:.1%} != expected {expected_pct:.1%}")
-            self.assertAlmostEqual(kosync_pct, expected_pct, delta=tolerance,
-                                   msg=f"KoSync final state {kosync_pct:.1%} != expected {expected_pct:.1%}")
+            if kosync_skipped:
+                self.assertNotIn('kosync', final_states, "KoSync state should not be persisted when malformed XPath update is skipped")
+            else:
+                self.assertAlmostEqual(kosync_pct, expected_pct, delta=tolerance,
+                                       msg=f"KoSync final state {kosync_pct:.1%} != expected {expected_pct:.1%}")
             if self.test_book.storyteller_uuid:
                 self.assertAlmostEqual(storyteller_pct, expected_pct, delta=tolerance,
                                     msg=f"Storyteller final state {storyteller_pct:.1%} != expected {expected_pct:.1%}")

tests/test_abs_unittest.py

@@ -6,10 +6,14 @@
 import unittest
 import sys
 from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
 # Add the project root to the path to resolve module imports
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
 from tests.base_sync_test import BaseSyncCycleTestCase
+from src.sync_clients.kosync_sync_client import KoSyncSyncClient
+from src.sync_clients.sync_client_interface import LocatorResult
 
 class TestABSLeadsSync(BaseSyncCycleTestCase):
     """Test case for ABS leading sync_cycle scenario."""
@@ -74,6 +78,28 @@ def get_progress_mock_returns(self):
     def test_abs_leads(self):
         super().run_test(10, 40)
 
+    def test_malformed_xpath_skips_kosync_update(self):
+        kosync_api = Mock()
+        ebook_parser = Mock()
+        ebook_parser.get_sentence_level_ko_xpath.return_value = None
+        client = KoSyncSyncClient(kosync_api, ebook_parser)
+
+        book = SimpleNamespace(
+            kosync_doc_id='test-kosync-doc',
+            ebook_filename='test-book.epub',
+            abs_title='Test Audiobook',
+        )
+        request = SimpleNamespace(
+            locator_result=LocatorResult(percentage=0.4, xpath="/html/body/div[1]/p[5]")
+        )
+
+        result = client.update_progress(book, request)
+
+        self.assertFalse(result.success)
+        self.assertTrue(result.updated_state.get('skipped'))
+        self.assertIsNone(result.updated_state.get('xpath'))
+        kosync_api.update_progress.assert_not_called()
+
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)

tests/test_ebook_sentence_xpath_fallback.py

@@ -43,7 +43,7 @@ def test_crengine_safe_xpath_falls_back_when_anchor_has_no_direct_text(self):
 
         xpath = self.parser._build_crengine_safe_text_xpath(span, 8, html_content)
 
-        self.assertEqual(xpath, "/body/DocFragment[8]/body/p[1]/text().0")
+        self.assertEqual(xpath, "/body/DocFragment[8]/body/p/text().0")
         self.assertNotIn("/span", xpath)