updates UI
This commit is contained in:
mike
@@ -623,5 +623,181 @@ class TestAPIRegression(unittest.TestCase):
|
||||
self.assertEqual(dance_grp["video"], "dance.mp4")
|
||||
self.assertIn("dance_10s-20s.mp4", dance_grp["clips"])
|
||||
|
||||
def test_12_standalone_scenery_sorting_and_existence_check(self):
|
||||
from unittest.mock import patch
|
||||
import edit_api
|
||||
import database
|
||||
|
||||
# Insert some mock scenery generation records so we can test last_used sorting
|
||||
conn = database.get_db_connection()
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
cur.execute("""
|
||||
INSERT INTO person (filename, group_id, source_refs)
|
||||
VALUES ('20260629_130000_sc_test.png', 'g1', '["test_person.png", "wireframe:used_scenery_2.png"]')
|
||||
""")
|
||||
cur.execute("""
|
||||
INSERT INTO person (filename, group_id, source_refs)
|
||||
VALUES ('20260629_130500_sc_test.png', 'g1', '["test_person.png", "wireframe:used_scenery_1.png"]')
|
||||
""")
|
||||
conn.commit()
|
||||
finally:
|
||||
cur.close()
|
||||
database._put_db_connection(conn)
|
||||
|
||||
try:
|
||||
# Mock wireframe folder and output directories
|
||||
with patch("edit_api._load_wireframe_dir", return_value="/tmp/test_wireframe"), \
|
||||
patch("edit_api._load_output_dir", return_value=self.output_dir), \
|
||||
patch("os.path.isdir", return_value=True), \
|
||||
patch("os.path.exists", return_value=True), \
|
||||
patch("os.path.getmtime", return_value=12345.0), \
|
||||
patch("os.listdir", return_value=["used_scenery_1.png", "used_scenery_2.png", "unused_scenery_3.png"]):
|
||||
|
||||
response = self.client.get("/videos?refresh=true")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
|
||||
# Check standalone images exist in the response
|
||||
self.assertIn("standalone_images", data)
|
||||
standalone = data["standalone_images"]
|
||||
|
||||
# Should contain our wireframes sorted with last used on top:
|
||||
# 'used_scenery_1.png' was used in the most recent scenery generation (at 13:05:00)
|
||||
# 'used_scenery_2.png' was used in the older scenery generation (at 13:00:00)
|
||||
# 'unused_scenery_3.png' was not used
|
||||
self.assertEqual(standalone[0], "used_scenery_1.png")
|
||||
self.assertEqual(standalone[1], "used_scenery_2.png")
|
||||
self.assertEqual(standalone[2], "unused_scenery_3.png")
|
||||
|
||||
# Also verify that /scenery/library filters out missing files
|
||||
# Let's mock _get_cached_file_meta: '20260629_130500_sc_test.png' exists, but '20260629_130000_sc_test.png' is missing
|
||||
def mock_cached_meta(fname, out_dir):
|
||||
if fname == "20260629_130500_sc_test.png":
|
||||
return True, 12345.0
|
||||
return False, 0.0
|
||||
|
||||
with patch("edit_api._get_cached_file_meta", side_effect=mock_cached_meta):
|
||||
response = self.client.get("/scenery/library")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
lib_data = response.json()
|
||||
|
||||
# Verify that only the existing scenery item '20260629_130500_sc_test.png' is returned
|
||||
all_filenames = []
|
||||
for g in lib_data["groups"]:
|
||||
for item in g["items"]:
|
||||
all_filenames.append(item["filename"])
|
||||
for item in lib_data["ungrouped"]:
|
||||
all_filenames.append(item["filename"])
|
||||
|
||||
self.assertIn("20260629_130500_sc_test.png", all_filenames)
|
||||
self.assertNotIn("20260629_130000_sc_test.png", all_filenames)
|
||||
|
||||
finally:
|
||||
# Clean up test DB rows
|
||||
conn = database.get_db_connection()
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
cur.execute("DELETE FROM person WHERE filename IN ('20260629_130000_sc_test.png', '20260629_130500_sc_test.png')")
|
||||
conn.commit()
|
||||
finally:
|
||||
cur.close()
|
||||
database._put_db_connection(conn)
|
||||
|
||||
def test_13_scenery_metadata_extraction_and_refs(self):
|
||||
from unittest.mock import patch, MagicMock
|
||||
import edit_api
|
||||
import database
|
||||
import json
|
||||
import os
|
||||
from PIL import Image
|
||||
|
||||
# Write dummy model image on disk
|
||||
model_filename = "model_image_123.png"
|
||||
model_path = os.path.join(self.output_dir, model_filename)
|
||||
Image.new("RGB", (10, 10), color="red").save(model_path)
|
||||
|
||||
# Set up a test person record in DB for the model
|
||||
conn = database.get_db_connection()
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
cur.execute("""
|
||||
INSERT INTO person (filename, group_id)
|
||||
VALUES (%s, 'model_group_123')
|
||||
""", (model_filename,))
|
||||
conn.commit()
|
||||
finally:
|
||||
cur.close()
|
||||
database._put_db_connection(conn)
|
||||
|
||||
try:
|
||||
# Let's mock _run_pipeline, generation of embeddings, etc.
|
||||
# We want to verify that _scenery_worker correctly structures 'source_refs'
|
||||
# and triggers metadata extraction.
|
||||
mock_png = b"dummy_png_bytes"
|
||||
edit_api.jobs["test_scenery_job"] = {"status": "running"}
|
||||
|
||||
with patch("edit_api._run_pipeline", return_value=mock_png), \
|
||||
patch("edit_api._load_output_dir", return_value=self.output_dir), \
|
||||
patch("edit_api.embeddings.generate_embedding", return_value=None), \
|
||||
patch("edit_api._metadata_executor.submit") as mock_submit:
|
||||
|
||||
# Run the scenery worker synchronously for the test
|
||||
dummy_scene = Image.new("RGB", (100, 100), color="blue")
|
||||
|
||||
edit_api._scenery_worker(
|
||||
job_id="test_scenery_job",
|
||||
model_filename=model_filename,
|
||||
scene_pil=dummy_scene,
|
||||
prompt="test scenery prompt",
|
||||
seed=42,
|
||||
extra_pils=[],
|
||||
scene_video="test_video.mp4",
|
||||
scene_image="test_image.png",
|
||||
extra_filename="test_extra.png"
|
||||
)
|
||||
|
||||
# Find the newly created scenery filename
|
||||
scenery_filename = edit_api.jobs["test_scenery_job"]["output"]
|
||||
self.assertIsNotNone(scenery_filename)
|
||||
|
||||
# Check database record to verify refs are correct
|
||||
person = database.get_person(scenery_filename)
|
||||
self.assertIsNotNone(person)
|
||||
self.assertEqual(person[1], "model_group_123") # Correctly pulled group_id
|
||||
|
||||
refs = json.loads(person[12]) # source_refs is at index 12 in tuple or get_person response
|
||||
self.assertIn(model_filename, refs)
|
||||
self.assertIn("video:test_video.mp4", refs)
|
||||
self.assertIn("wireframe:test_image.png", refs)
|
||||
self.assertIn("test_extra.png", refs)
|
||||
|
||||
# Verify that _metadata_executor.submit was called to trigger background metadata extraction
|
||||
mock_submit.assert_called_once_with(edit_api._process_image_for_metadata, scenery_filename)
|
||||
|
||||
finally:
|
||||
# Clean up all created files and database entries
|
||||
if os.path.exists(model_path):
|
||||
os.remove(model_path)
|
||||
conn = database.get_db_connection()
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
cur.execute("DELETE FROM person WHERE filename = 'model_image_123.png'")
|
||||
cur.execute("DELETE FROM person WHERE filename LIKE '%_sc_model_image_123.png'")
|
||||
conn.commit()
|
||||
finally:
|
||||
cur.close()
|
||||
database._put_db_connection(conn)
|
||||
|
||||
# Clean up generated scenery file on disk
|
||||
try:
|
||||
scenery_filename = edit_api.jobs.get("test_scenery_job", {}).get("output")
|
||||
if scenery_filename:
|
||||
out_path = os.path.join(self.output_dir, scenery_filename)
|
||||
if os.path.exists(out_path):
|
||||
os.remove(out_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user