-
Type: Bug
-
Status: Open
-
Priority: Minor
-
Resolution: Unresolved
-
Affects Version/s: None
-
Fix Version/s: None
-
Component/s: Tests
-
Story Points:2
The test is failing every time:
$ python -m pytest -n0 tests/old_functional/test_direct_transfer.py::TestDirectTransferFolder::test_pause_resume_session_non_chunked $ python -m pytest -n0 tests/old_functional/test_direct_transfer.py::TestDirectTransferFolderNoSync::test_pause_resume_session_non_chunked
def test_pause_resume_session_non_chunked(self): """ Test the session pause and resume system for sessions containing non-chunked files. The Session final status should be COMPLETED. """ engine = self.engine_1 # There is no upload, right now assert not list(engine.dao.get_dt_uploads()) expression = re.compile( r"<LogRecord: nxdrive\.notification, .*, .*, .*, " r"\"Sending Notification\(level='info' title='Direct Transfer'" r" uid='DIRECT_TRANSFER_SESSION_END.*' unique=False\)\">" ) upload_count = 0 def get_upload(*_, **__): """Alternative version of EngineDAO.get_upload() that pause the session.""" nonlocal upload_count # The first upload is the folder, we want to pause the session just before the file. if upload_count == 0: upload_count += 1 return None # Ensure we have 0 ongoing upload dao = engine.dao uploads = list(dao.get_dt_uploads()) assert not uploads # Verify the session status sessions = dao.get_active_sessions_raw() assert len(sessions) == 1 session = sessions[0] assert session["total"] == 2 assert session["status"] is TransferStatus.ONGOING # Pause the session dao.pause_session(session["uid"]) # Session should be paused now session = dao.get_session(session["uid"]) assert session.status is TransferStatus.PAUSED return None created = [] root_folder = self.tmpdir / str(uuid4())[:6] root_folder.mkdir() created.append(root_folder) sub_file = root_folder / f"file_{str(uuid4())[:4]}" sub_file.write_text("Some content.", encoding="utf8") created.append(sub_file) with patch.object(engine.dao, "get_upload", new=get_upload): with ensure_no_exception(): self.direct_transfer(root_folder) self.wait_sync() session = engine.dao.get_session(1) assert session assert session.status is TransferStatus.PAUSED uploads = list(engine.dao.get_dt_uploads()) > assert uploads E AssertionError: assert []
- Is referenced in