From 9773ef2b80bd692d0d8e2158073e3849b79d8580 Mon Sep 17 00:00:00 2001 From: CodeBeaver Date: Tue, 15 Apr 2025 11:28:31 +0000 Subject: [PATCH] codebeaver/pre/beta-963 - . --- tests/graphs/abstract_graph_test.py | 63 +++++ tests/test_chromium.py | 262 ++++++++++++++++++ tests/test_json_scraper_multi_graph.py | 0 tests/test_omni_search_graph.py | 110 ++++++++ tests/test_script_creator_multi_graph.py | 215 ++++++++++++++ .../test_smart_scraper_multi_concat_graph.py | 0 6 files changed, 650 insertions(+) create mode 100644 tests/test_json_scraper_multi_graph.py create mode 100644 tests/test_omni_search_graph.py create mode 100644 tests/test_script_creator_multi_graph.py create mode 100644 tests/test_smart_scraper_multi_concat_graph.py diff --git a/tests/graphs/abstract_graph_test.py b/tests/graphs/abstract_graph_test.py index 4c9b026e..43a71018 100644 --- a/tests/graphs/abstract_graph_test.py +++ b/tests/graphs/abstract_graph_test.py @@ -14,6 +14,67 @@ """ +def test_llm_missing_tokens(monkeypatch, capsys): + """Test that missing model tokens causes default to 8192 with an appropriate warning printed.""" + # Patch out models_tokens to simulate missing tokens for the given model + from scrapegraphai.graphs import abstract_graph + + monkeypatch.setattr( + abstract_graph, "models_tokens", {"openai": {"gpt-3.5-turbo": 4096}} + ) + llm_config = {"model": "openai/not-known-model", "openai_api_key": "test"} + # Patch _create_graph to return a dummy graph to avoid real graph creation + with patch.object(TestGraph, "_create_graph", return_value=Mock(nodes=[])): + graph = TestGraph("Test prompt", {"llm": llm_config}) + # Since "not-known-model" is missing, it should default to 8192 + assert graph.model_token == 8192 + captured = capsys.readouterr().out + assert "Max input tokens for model" in captured + + +def test_burr_kwargs(): + """Test that burr_kwargs configuration correctly sets use_burr and burr_config on the graph.""" + dummy_graph = Mock() + dummy_graph.nodes = [] + with patch.object(TestGraph, "_create_graph", return_value=dummy_graph): + config = { + "llm": {"model": "openai/gpt-3.5-turbo", "openai_api_key": "sk-test"}, + "burr_kwargs": {"some_key": "some_value"}, + } + graph = TestGraph("Test prompt", config) + # Check that the burr_kwargs have been applied and an app_instance_id added if missing + assert dummy_graph.use_burr is True + assert dummy_graph.burr_config["some_key"] == "some_value" + assert "app_instance_id" in dummy_graph.burr_config + + +def test_set_common_params(): + """ + Test that the set_common_params method correctly updates the configuration + of all nodes in the graph. + """ + # Create a mock graph with mock nodes + mock_graph = Mock() + mock_node1 = Mock() + mock_node2 = Mock() + mock_graph.nodes = [mock_node1, mock_node2] + # Create a TestGraph instance with the mock graph + with patch( + "scrapegraphai.graphs.abstract_graph.AbstractGraph._create_graph", + return_value=mock_graph, + ): + graph = TestGraph( + "Test prompt", + {"llm": {"model": "openai/gpt-3.5-turbo", "openai_api_key": "sk-test"}}, + ) + # Call set_common_params with test parameters + test_params = {"param1": "value1", "param2": "value2"} + graph.set_common_params(test_params) + # Assert that update_config was called on each node with the correct parameters + mock_node1.update_config.assert_called_once_with(test_params, False) + mock_node2.update_config.assert_called_once_with(test_params, False) + + class TestGraph(AbstractGraph): def __init__(self, prompt: str, config: dict): super().__init__(prompt, config) @@ -78,6 +139,7 @@ class TestAbstractGraph: { "model": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0", "region_name": "IDK", + "temperature": 0.7, }, ChatBedrock, ), @@ -136,6 +198,7 @@ def test_create_llm_unknown_provider(self): { "model": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0", "region_name": "IDK", + "temperature": 0.7, "rate_limit": {"requests_per_second": 1}, }, ChatBedrock, diff --git a/tests/test_chromium.py b/tests/test_chromium.py index e6d79b52..976a3cdd 100644 --- a/tests/test_chromium.py +++ b/tests/test_chromium.py @@ -1,5 +1,6 @@ import asyncio import sys +import time from unittest.mock import ANY, AsyncMock, patch import aiohttp @@ -1934,3 +1935,264 @@ def fake_parse_or_search_proxy(proxy): ) with pytest.raises(ValueError, match="Invalid proxy"): ChromiumLoader(["http://example.com"], backend="playwright", proxy="bad_proxy") + + +@pytest.mark.asyncio +async def test_alazy_load_with_single_url_string(monkeypatch): + """Test that alazy_load yields Document objects when urls is a string (iterating over characters).""" + # Passing a string as URL; lazy_load will iterate each character. + loader = ChromiumLoader( + "http://example.com", backend="playwright", requires_js_support=False + ) + + async def dummy_scraper(url, browser_name="chromium"): + return f"{url}" + + monkeypatch.setattr(loader, "ascrape_playwright", dummy_scraper) + docs = [doc async for doc in loader.alazy_load()] + # The expected number of documents is the length of the string + expected_length = len("http://example.com") + assert len(docs) == expected_length + # Check that the first document’s source is the first character ('h') + assert docs[0].metadata["source"] == "h" + + +def test_lazy_load_with_single_url_string(monkeypatch): + """Test that lazy_load yields Document objects when urls is a string (iterating over characters).""" + loader = ChromiumLoader( + "http://example.com", backend="playwright", requires_js_support=False + ) + + async def dummy_scraper(url, browser_name="chromium"): + return f"{url}" + + monkeypatch.setattr(loader, "ascrape_playwright", dummy_scraper) + docs = list(loader.lazy_load()) + expected_length = len("http://example.com") + assert len(docs) == expected_length + # The first character from the URL is 'h' + assert docs[0].metadata["source"] == "h" + + +@pytest.mark.asyncio +async def test_ascrape_playwright_scroll_invalid_type(monkeypatch): + """Test that ascrape_playwright_scroll raises TypeError when invalid types are passed for scroll or sleep.""" + # Create a dummy playwright so that evaluate and content can be called + + loader = ChromiumLoader(["http://example.com"], backend="playwright") + # Passing a non‐numeric sleep value should eventually trigger an error + with pytest.raises(TypeError): + await loader.ascrape_playwright_scroll( + "http://example.com", scroll=6000, sleep="2", scroll_to_bottom=False + ) + + +@pytest.mark.asyncio +async def test_alazy_load_non_iterable_urls(): + """Test that alazy_load raises TypeError when urls is not an iterable (e.g., integer).""" + with pytest.raises(TypeError): + # Passing an integer as urls should cause a TypeError during iteration. + loader = ChromiumLoader(123, backend="playwright") + [doc async for doc in loader.alazy_load()] + + +def test_lazy_load_non_iterable_urls(): + """Test that lazy_load raises TypeError when urls is not an iterable (e.g., integer).""" + with pytest.raises(TypeError): + loader = ChromiumLoader(456, backend="playwright") + + class DummyPW: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return + + class chromium: + @staticmethod + async def launch(headless, proxy, **kwargs): + return DummyBrowser() + + class firefox: + @staticmethod + async def launch(headless, proxy, **kwargs): + return DummyBrowser() + + monkeypatch.setattr("playwright.async_api.async_playwright", lambda: DummyPW()) + + # Create a loader instance with retry_limit=2 so that one failure is allowed. + + +@pytest.mark.asyncio +async def test_ascrape_playwright_caplog(monkeypatch, caplog): + """ + Test that ascrape_playwright recovers on failure and that error messages are logged. + This test simulates one failed attempt (via a Timeout) and then a successful attempt. + """ + # Create a loader instance with a retry limit of 2 and a short timeout. + loader = ChromiumLoader( + ["http://example.com"], backend="playwright", retry_limit=2, timeout=1 + ) + attempt = {"count": 0} + + async def dummy_ascrape(url, browser_name="chromium"): + if attempt["count"] < 1: + attempt["count"] += 1 + raise asyncio.TimeoutError("Simulated Timeout") + return "Recovered Content" + + monkeypatch.setattr(loader, "ascrape_playwright", dummy_ascrape) + with caplog.at_level("ERROR"): + result = await loader.ascrape_playwright("http://example.com") + assert "Recovered Content" in result + assert any( + "Attempt 1 failed: Simulated Timeout" in record.message + for record in caplog.records + ) + + class DummyContext: + def __init__(self): + self.new_page_called = False + + async def new_page(self): + self.new_page_called = True + return DummyPage() + + class DummyBrowser: + def __init__(self): + self.new_context_kwargs = None + + async def new_context(self, **kwargs): + self.new_context_kwargs = kwargs + return DummyContext() + + async def close(self): + return + + class DummyPW: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return + + class chromium: + @staticmethod + async def launch(headless, proxy, **kwargs): + return DummyBrowser() + + monkeypatch.setattr("playwright.async_api.async_playwright", lambda: DummyPW()) + + # Initialize the loader with a non-empty storage_state value. + loader = ChromiumLoader( + ["http://example.com"], backend="playwright", storage_state="dummy_state" + ) + + # Call ascrape_playwright and capture its result. + result = await loader.ascrape_playwright("http://example.com") + + # To verify that ignore_https_errors was passed into new_context, + # simulate a separate launch to inspect the new_context_kwargs. + browser_instance = await DummyPW.chromium.launch( + headless=loader.headless, proxy=loader.proxy + ) + await browser_instance.new_context( + storage_state=loader.storage_state, ignore_https_errors=True + ) + kwargs = browser_instance.new_context_kwargs + + assert kwargs is not None + assert kwargs.get("ignore_https_errors") is True + assert kwargs.get("storage_state") == "dummy_state" + assert "Ignore HTTPS errors Test" in result + + +@pytest.mark.asyncio +async def test_ascrape_with_js_support_context_error_cleanup(monkeypatch): + """Test that ascrape_with_js_support calls browser.close() even if new_context fails.""" + close_called = {"called": False} + + class DummyBrowser: + async def new_context(self, **kwargs): + # Force an exception during context creation + raise Exception("Context error") + + async def close(self): + close_called["called"] = True + + class DummyPW: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return + + class chromium: + @staticmethod + async def launch(headless, proxy, **kwargs): + return DummyBrowser() + + class firefox: + @staticmethod + async def launch(headless, proxy, **kwargs): + return DummyBrowser() + + monkeypatch.setattr("playwright.async_api.async_playwright", lambda: DummyPW()) + loader = ChromiumLoader( + ["http://example.com"], + backend="playwright", + requires_js_support=True, + retry_limit=1, + timeout=1, + ) + with pytest.raises(RuntimeError, match="Failed to scrape after 1 attempts"): + await loader.ascrape_with_js_support("http://example.com") + assert close_called["called"] is True + + +@pytest.mark.asyncio +async def test_lazy_load_with_none_urls(monkeypatch): + """Test that lazy_load raises TypeError when urls is None.""" + loader = ChromiumLoader(None, backend="playwright") + with pytest.raises(TypeError): + list(loader.lazy_load()) + + +@pytest.mark.asyncio +def test_lazy_load_sequential_timing(monkeypatch): + """Test that lazy_load runs scraping sequentially rather than concurrently.""" + urls = ["http://example.com/1", "http://example.com/2", "http://example.com/3"] + loader = ChromiumLoader(urls, backend="playwright", requires_js_support=False) + + async def dummy_scraper_with_delay(url, browser_name="chromium"): + await asyncio.sleep(0.5) + return f"Delayed content for {url}" + + monkeypatch.setattr(loader, "ascrape_playwright", dummy_scraper_with_delay) + start = time.monotonic() + docs = list(loader.lazy_load()) + elapsed = time.monotonic() - start + # At least 0.5 seconds per URL should be observed. + assert elapsed >= 1.5, ( + f"Sequential lazy_load took too little time: {elapsed:.2f} seconds" + ) + for doc, url in zip(docs, urls): + assert f"Delayed content for {url}" in doc.page_content + assert doc.metadata["source"] == url + + +@pytest.mark.asyncio +def test_lazy_load_with_tuple_urls(monkeypatch): + """Test that lazy_load yields Document objects correctly when urls is provided as a tuple.""" + urls = ("http://example.com", "http://test.com") + loader = ChromiumLoader(urls, backend="playwright", requires_js_support=False) + + async def dummy_scraper(url, browser_name="chromium"): + return f"Tuple content for {url}" + + monkeypatch.setattr(loader, "ascrape_playwright", dummy_scraper) + docs = list(loader.lazy_load()) + assert len(docs) == 2 + for doc, url in zip(docs, urls): + assert f"Tuple content for {url}" in doc.page_content + assert doc.metadata["source"] == url diff --git a/tests/test_json_scraper_multi_graph.py b/tests/test_json_scraper_multi_graph.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/test_omni_search_graph.py b/tests/test_omni_search_graph.py new file mode 100644 index 00000000..656421d5 --- /dev/null +++ b/tests/test_omni_search_graph.py @@ -0,0 +1,110 @@ +from pydantic import BaseModel + +# Import the class under test +from scrapegraphai.graphs.omni_search_graph import OmniSearchGraph + + +# Create a dummy graph class to simulate graph execution +class DummyGraph: + def __init__(self, final_state): + self.final_state = final_state + + def execute(self, inputs): + # Return final_state and dummy execution info + return self.final_state, {"debug": True} + + +# Dummy schema for testing purposes +class DummySchema(BaseModel): + result: str + + +class TestOmniSearchGraph: + """Test suite for the OmniSearchGraph module.""" + + def test_run_with_answer(self): + """Test that the run() method returns the correct answer when present.""" + config = { + "llm": {"model": "dummy-model"}, + "max_results": 3, + "search_engine": "dummy-engine", + } + prompt = "Test prompt?" + graph_instance = OmniSearchGraph(prompt, config) + # Set required attribute manually + graph_instance.llm_model = {"model": "dummy-model"} + # Inject a DummyGraph that returns a final state containing an "answer" + dummy_final_state = {"answer": "expected answer"} + graph_instance.graph = DummyGraph(dummy_final_state) + result = graph_instance.run() + assert result == "expected answer" + + def test_run_without_answer(self): + """Test that the run() method returns the default message when no answer is found.""" + config = { + "llm": {"model": "dummy-model"}, + "max_results": 3, + "search_engine": "dummy-engine", + } + prompt = "Test prompt without answer?" + graph_instance = OmniSearchGraph(prompt, config) + graph_instance.llm_model = {"model": "dummy-model"} + # Inject a DummyGraph that returns an empty final state + dummy_final_state = {} + graph_instance.graph = DummyGraph(dummy_final_state) + result = graph_instance.run() + assert result == "No answer found." + + def test_create_graph_structure(self): + """Test that the _create_graph() method returns a graph with the expected structure.""" + config = { + "llm": {"model": "dummy-model"}, + "max_results": 4, + "search_engine": "dummy-engine", + } + prompt = "Structure test prompt" + # Using a dummy schema for testing + graph_instance = OmniSearchGraph(prompt, config, schema=DummySchema) + graph_instance.llm_model = {"model": "dummy-model"} + constructed_graph = graph_instance._create_graph() + # Ensure constructed_graph has essential attributes + assert hasattr(constructed_graph, "nodes") + assert hasattr(constructed_graph, "edges") + assert hasattr(constructed_graph, "entry_point") + assert hasattr(constructed_graph, "graph_name") + # Check that the graph_name matches the class name + assert constructed_graph.graph_name == "OmniSearchGraph" + # Expecting three nodes and two edges as per the implementation + assert len(constructed_graph.nodes) == 3 + assert len(constructed_graph.edges) == 2 + + def test_config_deepcopy(self): + """Test that the config passed to OmniSearchGraph is deep copied properly.""" + config = { + "llm": {"model": "dummy-model"}, + "max_results": 2, + "search_engine": "dummy-engine", + } + prompt = "Deepcopy test" + graph_instance = OmniSearchGraph(prompt, config) + graph_instance.llm_model = {"model": "dummy-model"} + # Modify the original config after instantiation + config["llm"]["model"] = "changed-model" + # The internal copy should remain unchanged + assert graph_instance.copy_config["llm"]["model"] == "dummy-model" + + def test_schema_deepcopy(self): + """Test that the schema is deep copied correctly so external changes do not affect it.""" + config = { + "llm": {"model": "dummy-model"}, + "max_results": 2, + "search_engine": "dummy-engine", + } + # Instantiate with DummySchema + graph_instance = OmniSearchGraph("Schema test", config, schema=DummySchema) + graph_instance.llm_model = {"model": "dummy-model"} + # Modify the internal copy of the schema directly to simulate isolation + graph_instance.copy_schema = DummySchema(result="internal") + external_schema = DummySchema(result="external") + external_schema.result = "modified" + assert graph_instance.copy_schema.result == "internal" diff --git a/tests/test_script_creator_multi_graph.py b/tests/test_script_creator_multi_graph.py new file mode 100644 index 00000000..3341c88a --- /dev/null +++ b/tests/test_script_creator_multi_graph.py @@ -0,0 +1,215 @@ +import pytest +from pydantic import BaseModel + +from scrapegraphai.graphs.script_creator_graph import ScriptCreatorGraph +from scrapegraphai.graphs.script_creator_multi_graph import ( + BaseGraph, + ScriptCreatorMultiGraph, +) + + +@pytest.fixture(autouse=True) +def set_api_key_env(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "dummy") + + +# Dummy classes to simulate behavior for testing +class DummyGraph: + def __init__(self, final_state, execution_info): + self.final_state = final_state + self.execution_info = execution_info + + def execute(self, inputs): + return self.final_state, self.execution_info + + +class DummySchema(BaseModel): + field: str = "dummy" + + +class TestScriptCreatorMultiGraph: + """Tests for ScriptCreatorMultiGraph.""" + + def test_run_success(self): + """Test run() returns the merged script when execution is successful.""" + prompt = "Test prompt" + source = ["http://example.com"] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + # Set necessary attributes that are expected by _create_graph() and the run() method. + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + # Replace the graph with a dummy graph that simulates successful execution. + dummy_final_state = {"merged_script": "print('Hello World')"} + dummy_execution_info = {"info": "dummy"} + instance.graph = DummyGraph(dummy_final_state, dummy_execution_info) + result = instance.run() + assert result == "print('Hello World')" + + def test_run_failure(self): + """Test run() returns failure message when merged_script is missing.""" + prompt = "Test prompt" + source = ["http://example.com"] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + dummy_final_state = {"other_key": "no script"} + dummy_execution_info = {"info": "dummy"} + instance.graph = DummyGraph(dummy_final_state, dummy_execution_info) + result = instance.run() + assert result == "Failed to generate the script." + + def test_create_graph_structure(self): + """Test _create_graph() returns a BaseGraph with the correct graph name and structure.""" + prompt = "Test prompt" + source = [] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + # Manually assign llm_model and schema for node configuration in the graph. + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + graph = instance._create_graph() + assert isinstance(graph, BaseGraph) + assert hasattr(graph, "graph_name") + assert graph.graph_name == "ScriptCreatorMultiGraph" + # Check that the graph has two nodes. + assert len(graph.nodes) == 2 + # Optional: Check that the edges list is correctly formed. + assert len(graph.edges) == 1 + + def test_config_deepcopy(self): + """Test that the configuration is deep copied during initialization.""" + prompt = "Test prompt" + source = [] + config = {"llm": {"model": "openai/test-model"}, "other": [1, 2, 3]} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + # Modify the original config. + config["llm"]["model"] = "changed-model" + config["other"].append(4) + # Verify that the config copied within instance remains unchanged. + assert instance.copy_config["llm"]["model"] == "openai/test-model" + assert instance.copy_config["other"] == [1, 2, 3] + + def test_init_attributes(self): + """Test that initial attributes are set correctly upon initialization.""" + prompt = "Initialization test" + source = ["http://init.com"] + config = {"llm": {"model": "openai/init-model"}, "param": [1, 2]} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + # Check that basic attributes are set correctly + assert instance.prompt == prompt + assert instance.source == source + # Check that copy_config is a deep copy and equals the original config + assert instance.copy_config == { + "llm": {"model": "openai/init-model"}, + "param": [1, 2], + } + # For classes, deepcopy returns the same object, so the copy_schema should equal schema + assert instance.copy_schema == DummySchema + + def test_run_no_schema(self): + """Test run() when schema is None.""" + prompt = "No schema prompt" + source = ["http://noschema.com"] + config = {"llm": {"model": "openai/no-schema"}} + instance = ScriptCreatorMultiGraph(prompt, source, config, schema=None) + instance.llm_model = {"model": "openai/no-schema"} + instance.schema = None + dummy_final_state = {"merged_script": "print('No Schema Script')"} + dummy_execution_info = {"info": "no schema"} + instance.graph = DummyGraph(dummy_final_state, dummy_execution_info) + result = instance.run() + assert result == "print('No Schema Script')" + + def test_create_graph_node_configs(self): + """Test that _create_graph() sets correct node configurations for its nodes.""" + prompt = "Graph config test" + source = ["http://graphconfig.com"] + config = {"llm": {"model": "openai/graph-model"}, "extra": [10]} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + # Manually assign llm_model and schema for node configuration + instance.llm_model = {"model": "openai/graph-model"} + instance.schema = {"type": "graph-dummy"} + graph = instance._create_graph() + # Validate configuration of the first node (GraphIteratorNode) + node1 = graph.nodes[0] + assert node1.node_config["graph_instance"] == ScriptCreatorGraph + assert node1.node_config["scraper_config"] == instance.copy_config + # Validate configuration of the second node (MergeGeneratedScriptsNode) + node2 = graph.nodes[1] + assert node2.node_config["llm_model"] == instance.llm_model + assert node2.node_config["schema"] == instance.schema + + def test_entry_point_node(self): + """Test that the graph entry point is the GraphIteratorNode (the first node).""" + prompt = "Entry point test" + source = ["http://entrypoint.com"] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + graph = instance._create_graph() + assert graph.entry_point == graph.nodes[0] + + def test_run_exception(self): + """Test that run() propagates exceptions raised by graph.execute.""" + prompt = "Exception test" + source = ["http://exception.com"] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + + # Create a dummy graph that raises an exception when execute is called. + class ExceptionGraph: + def execute(self, inputs): + raise ValueError("Testing exception") + + instance.graph = ExceptionGraph() + with pytest.raises(ValueError, match="Testing exception"): + instance.run() + + def test_run_with_empty_prompt(self): + """Test run() method with an empty prompt.""" + prompt = "" + source = ["http://emptyprompt.com"] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + dummy_final_state = {"merged_script": "print('Empty prompt')"} + dummy_execution_info = {"info": "empty prompt"} + instance.graph = DummyGraph(dummy_final_state, dummy_execution_info) + result = instance.run() + assert result == "print('Empty prompt')" + + def test_run_called_twice(self): + """Test that running run() twice returns consistent and updated results.""" + prompt = "Twice test" + source = ["http://twicetest.com"] + config = {"llm": {"model": "openai/test-model"}} + schema = DummySchema + instance = ScriptCreatorMultiGraph(prompt, source, config, schema) + instance.llm_model = {"model": "openai/test-model"} + instance.schema = {"type": "dummy"} + dummy_final_state = {"merged_script": "print('First run')"} + dummy_execution_info = {"info": "first run"} + dummy_graph = DummyGraph(dummy_final_state, dummy_execution_info) + instance.graph = dummy_graph + result1 = instance.run() + # Modify dummy graph's state for the second run. + dummy_graph.final_state["merged_script"] = "print('Second run')" + dummy_graph.execution_info = {"info": "second run"} + result2 = instance.run() + assert result1 == "print('First run')" + assert result2 == "print('Second run')" diff --git a/tests/test_smart_scraper_multi_concat_graph.py b/tests/test_smart_scraper_multi_concat_graph.py new file mode 100644 index 00000000..e69de29b