Spaces:
Running
Running
| """Turing can BUILD now, not just describe building. | |
| `dee/core/cloning.py` has done Gibson, Golden Gate and restriction cloning | |
| since the plasmid work — with junction detection, overhang checking and primer | |
| design — and until 2026-08-04 no agent tool imported it. So the "Build" step of | |
| this product's own Design→Build→Edit→Learn loop had no agent path: asked to | |
| clone three fragments into pET-28a, Turing could only write prose about how one | |
| would. The capability was in the repo the whole time. | |
| That is the fourth built-but-unreachable capability found in one day (the | |
| vector catalogue, BLAST, pairwise alignment, and this), which is why the | |
| capability audit leads with the category rather than with any one gap. | |
| The tests here are weighted toward the ways an assembly can LIE: | |
| * a design that does not close must be reported as a failure, not dressed up | |
| as a product; | |
| * a junction with no native overlap means primers someone has to buy, and a | |
| summary that omits that makes a design look free; | |
| * the engine's own warnings must survive to the model verbatim, not be | |
| summarised into optimism. | |
| """ | |
| import pytest | |
| from dee.core import agent_tools as t | |
| from dee.core import cloning as cl | |
| from dee.core import orchestrator as orch | |
| # Three fragments with real, detectable overlaps between neighbours. | |
| A = "ATGGCTAGCAAAGGAGAAGAACTTTTCACTGGAGTTGTCCCAATTCTTGTTGAATTAGATG" | |
| B = A[-30:] + "GTGATGTTAATGGGCACAAATTTTCTGTCAGTGGAGAGGGTGAAGGTGATGCAACATACG" | |
| C = B[-30:] + "GAAAACTTACCCTTAAATTTATTTGCACTACTGGAAAACTACCTGTTCCATGGCCAACAC" | |
| def _frags(): | |
| return [{"name": "vector", "sequence": A}, | |
| {"name": "insert1", "sequence": B}, | |
| {"name": "insert2", "sequence": C}] | |
| def _run(**kw): | |
| args = {"fragments": _frags()} | |
| args.update(kw) | |
| return t.execute_tool("simulate_assembly", args, auth_anonymous=True) | |
| # --------------------------------------------------------------------------- # | |
| # it actually assembles | |
| # --------------------------------------------------------------------------- # | |
| def test_the_agent_can_reach_the_cloning_engine_at_all(): | |
| """The whole point. Before this, cloning.py was not imported anywhere in | |
| the tool layer.""" | |
| assert "simulate_assembly" in t._TOOLS | |
| src = open("dee/core/agent_tools.py", encoding="utf-8").read() | |
| assert "from dee.core import cloning" in src | |
| def test_a_gibson_assembly_returns_a_real_product(): | |
| out = _run() | |
| assert out["ok"] is True | |
| assert out["_ui"]["sequence"] | |
| assert out["length"] == len(out["_ui"]["sequence"]) | |
| assert len(out["junctions"]) >= 2 | |
| def test_native_overlaps_are_detected_rather_than_designed_over(): | |
| """These fragments genuinely share 30 bp with their neighbours. If the | |
| engine reports them as 'designed', it is inventing primer work that the | |
| user does not need to pay for.""" | |
| out = _run() | |
| kinds = [j["kind"] for j in out["junctions"]] | |
| assert "native" in kinds, kinds | |
| def test_the_assembled_product_matches_the_engine_directly(): | |
| """The tool must not transform the sequence on its way through — it is a | |
| wrapper, and a wrapper that alters its payload is a second implementation.""" | |
| direct = cl.gibson([{"name": f["name"], "seq": f["sequence"]} for f in _frags()]) | |
| assert _run()["_ui"]["sequence"] == direct["assembled"] | |
| # --------------------------------------------------------------------------- # | |
| # the ways it can lie | |
| # --------------------------------------------------------------------------- # | |
| def test_a_failed_assembly_is_reported_as_a_failure(): | |
| """Golden Gate on fragments with no compatible overhangs does not close. | |
| That is a real, useful answer — and the one an agent is most tempted to | |
| round up into 'assembled successfully'.""" | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": _frags(), "method": "golden_gate"}, | |
| auth_anonymous=True) | |
| assert out["ok"] is False | |
| assert out["kind"] == "assembly_failed" | |
| assert "do not describe the product as if it assembled" in out["next"].lower() | |
| def test_primers_that_must_be_ordered_are_returned(): | |
| """A designed junction is a purchase. Returning the product without the | |
| primers makes the design look cheaper than it is.""" | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": [{"name": "x", "sequence": "ATGC" * 30}, | |
| {"name": "y", "sequence": "GGTT" * 30}]}, | |
| auth_anonymous=True) | |
| if out["ok"]: | |
| assert out["designed_junctions"] >= 1 | |
| assert "primers" in out | |
| def test_engine_warnings_survive_verbatim(): | |
| """Not summarised, not dropped. The warning is the honest part.""" | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": [{"name": "x", "sequence": "ATGC" * 30}, | |
| {"name": "y", "sequence": "GGTT" * 30}]}, | |
| auth_anonymous=True) | |
| if out["ok"]: | |
| direct = cl.gibson([{"name": "x", "seq": "ATGC" * 30}, | |
| {"name": "y", "seq": "GGTT" * 30}]) | |
| assert out["warnings"] == (direct.get("warnings") or []) | |
| def test_the_summary_leads_with_the_primer_cost(): | |
| """A one-line summary reading only "3,200 bp" hides the thing the user | |
| has to act on.""" | |
| out = _run() | |
| line = orch._summarize("simulate_assembly", out) | |
| assert "junction" in line | |
| if out.get("designed_junctions"): | |
| assert "primer" in line | |
| def test_the_result_asks_for_verification(): | |
| out = _run() | |
| assert "map_plasmid" in out["next"] | |
| assert "designed" in out["next"] | |
| # --------------------------------------------------------------------------- # | |
| # input handling | |
| # --------------------------------------------------------------------------- # | |
| def test_one_fragment_is_not_an_assembly(): | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": [{"name": "a", "sequence": A}]}, | |
| auth_anonymous=True) | |
| assert out["ok"] is False | |
| assert "two fragments" in out["error"] | |
| def test_a_fragment_with_no_dna_is_refused_by_name(): | |
| """Naming which fragment is empty is the difference between a fixable | |
| error and a shrug.""" | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": [{"name": "vector", "sequence": A}, | |
| {"name": "bad insert", "sequence": "???"}]}, | |
| auth_anonymous=True) | |
| assert out["ok"] is False | |
| assert "bad insert" in out["error"] | |
| def test_restriction_cloning_requires_its_enzyme(): | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": _frags(), "method": "restriction"}, | |
| auth_anonymous=True) | |
| assert out["ok"] is False and "enzyme" in out["error"] | |
| def test_an_unknown_method_is_refused_with_the_real_options(): | |
| out = t.execute_tool("simulate_assembly", | |
| {"fragments": _frags(), "method": "magic"}, | |
| auth_anonymous=True) | |
| assert out["ok"] is False | |
| for m in ("gibson", "golden_gate", "restriction"): | |
| assert m in out["error"] | |
| # --------------------------------------------------------------------------- # | |
| # wiring | |
| # --------------------------------------------------------------------------- # | |
| def test_the_bases_do_not_travel_through_the_model(): | |
| out = _run() | |
| stripped = orch._strip_ui(out) | |
| assert "_ui" not in stripped | |
| assert not any(isinstance(v, str) and len(v) > 40 and set(v) <= set("ACGT") | |
| for v in stripped.values()) | |
| def test_it_paints_the_plasmid_view(): | |
| assert orch._tool_view("simulate_assembly") == "plasmid" | |
| assert "simulate_assembly" in open("dee/static/cockpit.js", encoding="utf-8").read() | |
| def test_the_prompt_tells_it_to_build_rather_than_describe(): | |
| p = orch.build_system_prompt(False, {}, "") | |
| assert "BUILD MEANS BUILD" in p | |
| assert "simulate_assembly" in p | |
| # --------------------------------------------------------------------------- # | |
| # yeast in-vivo (homologous recombination) assembly | |
| # --------------------------------------------------------------------------- # | |
| def test_yeast_hr_returns_verification_marker_and_plan(): | |
| out = _run(method="yeast_hr", marker="URA3") | |
| assert out["ok"] is True and out["method"] == "yeast_hr" | |
| assert out["_ui"]["sequence"] | |
| assert out["marker"] == "URA3" | |
| assert set(out["plan"]) == {"transform", "selection", "verify"} | |
| # one colony-PCR verification pair per junction, both primers present | |
| assert len(out["verification"]) == len(out["junctions"]) | |
| assert all(v["forward"] and v["reverse"] for v in out["verification"]) | |
| assert all("unique" in j for j in out["junctions"]) | |
| def test_yeast_hr_flags_a_repeated_part_as_misassembly_risk(): | |
| import random | |
| rng = random.Random(12) | |
| def d(n): | |
| return "".join(rng.choice("ACGT") for _ in range(n)) | |
| term = d(45) # a terminator reused on two genes | |
| frags = [{"name": "vec HIS3", "sequence": d(220)}, | |
| {"name": "geneA-tADH1", "sequence": d(150) + term}, | |
| {"name": "geneB-tADH1", "sequence": d(150) + term}] | |
| out = t.execute_tool("simulate_assembly", | |
| {"method": "yeast_hr", "fragments": frags}, auth_anonymous=True) | |
| assert out["ok"] is True | |
| assert sum(1 for j in out["junctions"] if j.get("unique") is False) >= 2 | |
| assert any("share homology" in w for w in out["warnings"]) | |
| # --------------------------------------------------------------------------- # | |
| # yeast chromosomal integration (knock-in / knock-out) | |
| # --------------------------------------------------------------------------- # | |
| def _integ(**kw): | |
| import random | |
| rng = random.Random(31) | |
| def d(n): | |
| return "".join(rng.choice("ACGT") for _ in range(n)) | |
| args = {"upstream": d(300), "downstream": d(300), "insert": d(1200), "marker": "KanMX"} | |
| args.update(kw) | |
| return t.execute_tool("design_integration", args, auth_anonymous=True) | |
| def test_design_integration_returns_primers_verification_marker_and_plan(): | |
| out = _integ() | |
| assert out["ok"] is True and out["method"] == "yeast_integration" | |
| assert out["_ui"]["sequence"] and out["_ui"]["topology"] == "linear" | |
| assert out["marker"] == "KanMX" | |
| # cassette primers someone has to order are returned, never withheld | |
| assert out["primers"] and out["primers"][0]["forward"] and out["primers"][0]["reverse"] | |
| # one junction-PCR pair per side, both primers present | |
| assert len(out["verification"]) == 2 | |
| assert all(v["forward"] and v["reverse"] for v in out["verification"]) | |
| assert set(out["plan"]) == {"transform", "selection", "verify"} | |
| # the honest, genome-blind caveat rides all the way out to the model | |
| assert "BLAST" in out["next"] | |
| def test_design_integration_paints_the_build_card(): | |
| """The cockpit renders a result card off _ui.build — not warnings-as-prose. | |
| The card must carry the method, both junctions, the primers, the | |
| verification pairs, the plan and the BLAST caveat.""" | |
| b = _integ()["_ui"]["build"] | |
| assert b["method"] == "yeast_integration" | |
| assert b["primers"] and b["verification"] and b["plan"] | |
| assert len(b["junctions"]) == 2 | |
| assert any("BLAST" in w for w in b["warnings"]) | |
| def test_design_integration_requires_all_three_parts(): | |
| out = t.execute_tool("design_integration", | |
| {"upstream": "ACGT" * 20, "downstream": "TTGG" * 20}, | |
| auth_anonymous=True) | |
| assert out["ok"] is False | |
| assert "insert" in out["error"] or "cassette" in out["error"] | |
| def test_design_integration_cassette_primer_carries_a_homology_tail(): | |
| p = _integ()["primers"][0] | |
| assert p["forward_tail"] > 0 and p["reverse_tail"] > 0 | |
| assert p["forward_len"] == len(p["forward"]) | |
| def test_design_integration_paints_the_plasmid_view_and_names_the_blast(): | |
| out = _integ() | |
| assert orch._tool_view("design_integration") == "plasmid" | |
| line = orch._summarize("design_integration", out) | |
| assert "allele" in line | |
| assert "BLAST" in line or "ambiguous" in line | |
| def test_design_integration_bases_do_not_travel_through_the_model(): | |
| stripped = orch._strip_ui(_integ()) | |
| assert "_ui" not in stripped | |
| # the full integrated allele is only in the stripped _ui, never top-level | |
| assert not any(isinstance(v, str) and len(v) > 40 and set(v) <= set("ACGTN") | |
| for v in stripped.values()) | |