diff --git a/tests/test_web.py b/tests/test_web.py index f54b885..0fdd6f0 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -12,10 +12,13 @@ import dont_fret.web.state as state from dont_fret import PhotonData, PhotonFile +from dont_fret.config import cfg from dont_fret.web.bursts.components import BurstFigure +from dont_fret.web.datamanager import ThreadedDataManager from dont_fret.web.home.methods import task_burst_search -from dont_fret.web.methods import burst_search, create_file_items -from dont_fret.web.models import BurstNode, FRETNode, PhotonNode + +# from dont_fret.web.methods import burst_search, create_file_items +from dont_fret.web.models import BurstNode, FRETNode, ListStore, PhotonNode cwd = Path(__file__).parent input_data_dir = cwd / "test_data" / "input" @@ -31,39 +34,37 @@ def ph_ds1() -> PhotonData: @pytest.fixture -def photon_file_items() -> List[PhotonNode]: - file_items = create_file_items(input_data_dir / "ds1") - return file_items +def photon_nodes() -> List[PhotonNode]: + pth = input_data_dir / "ds1" + return [PhotonNode(file_path=ptu_pth) for ptu_pth in pth.glob("*.ptu")] @pytest.fixture -def burst_items(photon_file_items: List[PhotonNode]) -> List[BurstNode]: +def burst_nodes(photon_nodes: List[PhotonNode]) -> List[BurstNode]: + dm = ThreadedDataManager() burst_settings = ["DCBS", "APBS"] - dtype = pl.Enum([item.name for item in photon_file_items]) - burst_items = [] + burst_nodes = [] for name in burst_settings: - # TODO there should be a function list[PhotonFileItem] -> BurstItem - dfs = [burst_search(f, name, dtype) for f in photon_file_items] - df = pl.concat(dfs, how="vertical_relaxed") + node = asyncio.run(dm.get_burst_node(photon_nodes, cfg.burst_search[name], name=name)) + burst_nodes.append(node) - burst_items.append(BurstNode(name=name, df=df)) - return burst_items + return burst_nodes @pytest.mark.asyncio -async def test_burst_search(ph_ds1): +async def test_burst_search(): ph_item = PhotonNode(file_path=input_data_dir / "ds1" / "datafile_1.ptu") node = FRETNode( - name="FRET NOT", - photons=[ph_item], + name=solara.Reactive("FRET NOT"), + photons=ListStore([ph_item]), ) state.fret_nodes.set([]) - assert len(state.fret_nodes.value) == 0 + assert len(state.fret_nodes) == 0 state.fret_nodes.append(node) - assert len(state.fret_nodes.value) == 1 + assert len(state.fret_nodes) == 1 await task_burst_search.function("DCBS", node.id) # type: ignore @@ -77,21 +78,23 @@ async def test_burst_search(ph_ds1): await asyncio.sleep(0) -def test_burst_figure(photon_file_items, burst_items): - state.fret_nodes.set([]) +def test_burst_figure(photon_nodes, burst_nodes): node = FRETNode( - name="FRET NOT", - photons=photon_file_items, - bursts=burst_items, + name=solara.Reactive("FRET NOT"), + photons=ListStore(photon_nodes), + bursts=ListStore(burst_nodes), ) - state.fret_nodes.append(node) - fig = BurstFigure(state.fret_nodes, state.filters) + fig = BurstFigure([node]) box, rc = solara.render(fig) - locator = rc.find(v.Select, label="Burst item") - assert locator.widget.v_model == 0 - assert locator.widget.items == [{"text": "DCBS", "value": 0}, {"text": "APBS", "value": 1}] + locator = rc.find(v.Select, label="Bursts") + locator.wait_for(timeout=2.5) + assert locator.widget.v_model == burst_nodes[0].id.hex + assert locator.widget.items == [ + {"text": "DCBS", "value": burst_nodes[0].id.hex}, + {"text": "APBS", "value": burst_nodes[1].id.hex}, + ] find_figure = rc.find(go.FigureWidget).wait_for(timeout=5) find_figure.assert_single() @@ -99,7 +102,7 @@ def test_burst_figure(photon_file_items, burst_items): img_ref = np.load(output_data_dir / "ds1" / "z_img_dcbs.npy") assert np.allclose(img_test, img_ref) - locator.widget.v_model = 1 + locator.widget.v_model = burst_nodes[1].id.hex time.sleep(0.5) # wait for the redraw to start find_figure = rc.find(go.FigureWidget).wait_for(timeout=5) find_figure.assert_single()