diff --git a/pytest.ini b/pytest.ini index ae8b842b..f1e4f0a9 100644 --- a/pytest.ini +++ b/pytest.ini @@ -3,4 +3,5 @@ testpaths = tests markers = slow: slow tests git: tests that require git -addopts = --ignore=prometheus/script/ \ No newline at end of file +addopts = --ignore=prometheus/script/ +asyncio_mode = auto \ No newline at end of file diff --git a/tests/app/services/test_issue_service.py b/tests/app/services/test_issue_service.py index 9867f4d2..86242e51 100644 --- a/tests/app/services/test_issue_service.py +++ b/tests/app/services/test_issue_service.py @@ -44,7 +44,6 @@ def issue_service(mock_neo4j_service, mock_llm_service, mock_repository_service) ) -@pytest.mark.asyncio async def test_answer_issue_with_general_container(issue_service, monkeypatch): # Setup mock_issue_graph = Mock() @@ -106,7 +105,6 @@ async def test_answer_issue_with_general_container(issue_service, monkeypatch): assert result == ("test_patch", True, True, True, "test_response", IssueType.BUG) -@pytest.mark.asyncio async def test_answer_issue_with_user_defined_container(issue_service, monkeypatch): # Setup mock_issue_graph = Mock() diff --git a/tests/app/services/test_knowledge_graph_service.py b/tests/app/services/test_knowledge_graph_service.py index 202c73c2..8a64d3a0 100644 --- a/tests/app/services/test_knowledge_graph_service.py +++ b/tests/app/services/test_knowledge_graph_service.py @@ -41,7 +41,6 @@ def knowledge_graph_service(mock_neo4j_service, mock_kg_handler): return knowledge_graph_service -@pytest.mark.asyncio async def test_build_and_save_knowledge_graph(knowledge_graph_service, mock_kg_handler): """Test the build_and_save_knowledge_graph method.""" # Given diff --git a/tests/app/services/test_neo4j_service.py b/tests/app/services/test_neo4j_service.py index 0dd58b27..f64b8b99 100644 --- a/tests/app/services/test_neo4j_service.py +++ b/tests/app/services/test_neo4j_service.py @@ -6,12 +6,12 @@ @pytest.mark.slow async def test_neo4j_service(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - neo4j_service = Neo4jService( - neo4j_container.get_connection_url(), neo4j_container.username, neo4j_container.password - ) - assert neo4j_service.neo4j_driver is not None - try: - neo4j_service.neo4j_driver.verify_connectivity() - except Exception as e: - pytest.fail(f"Connection verification failed: {e}") + neo4j_container, kg = neo4j_container_with_kg_fixture + neo4j_service = Neo4jService( + neo4j_container.get_connection_url(), neo4j_container.username, neo4j_container.password + ) + assert neo4j_service.neo4j_driver is not None + try: + neo4j_service.neo4j_driver.verify_connectivity() + except Exception as e: + pytest.fail(f"Connection verification failed: {e}") diff --git a/tests/graph/test_knowledge_graph.py b/tests/graph/test_knowledge_graph.py index dff1dea8..5f6ddd35 100644 --- a/tests/graph/test_knowledge_graph.py +++ b/tests/graph/test_knowledge_graph.py @@ -74,9 +74,9 @@ async def test_get_file_tree_depth_two_max_seven_lines(): @pytest.mark.slow async def test_from_neo4j(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - driver = neo4j_container.get_driver() - handler = KnowledgeGraphHandler(driver, 100) - read_kg = handler.read_knowledge_graph(0, 1000, 100, 10) + neo4j_container, kg = neo4j_container_with_kg_fixture + driver = neo4j_container.get_driver() + handler = KnowledgeGraphHandler(driver, 100) + read_kg = handler.read_knowledge_graph(0, 1000, 100, 10) - assert read_kg == kg + assert read_kg == kg diff --git a/tests/lang_graph/nodes/test_context_provider_node.py b/tests/lang_graph/nodes/test_context_provider_node.py index cf15a5cb..48ef34f9 100644 --- a/tests/lang_graph/nodes/test_context_provider_node.py +++ b/tests/lang_graph/nodes/test_context_provider_node.py @@ -9,29 +9,27 @@ @pytest.mark.slow async def test_context_provider_node_basic_query(neo4j_container_with_kg_fixture): # noqa: F811 """Test basic query handling with the ContextProviderNode.""" - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - fake_response = "Fake response" - fake_llm = FakeListChatWithToolsModel(responses=[fake_response]) - node = ContextProviderNode( - model=fake_llm, - kg=kg, - neo4j_driver=neo4j_container.get_driver(), - max_token_per_result=1000, - ) + neo4j_container, kg = neo4j_container_with_kg_fixture + fake_response = "Fake response" + fake_llm = FakeListChatWithToolsModel(responses=[fake_response]) + node = ContextProviderNode( + model=fake_llm, + kg=kg, + neo4j_driver=neo4j_container.get_driver(), + max_token_per_result=1000, + ) - test_messages = [ - AIMessage(content="This code handles file processing"), - ToolMessage( - content="Found implementation in utils.py", tool_call_id="test_tool_call_1" - ), - ] - test_state = { - "original_query": "How does the error handling work?", - "context_provider_messages": test_messages, - } + test_messages = [ + AIMessage(content="This code handles file processing"), + ToolMessage(content="Found implementation in utils.py", tool_call_id="test_tool_call_1"), + ] + test_state = { + "original_query": "How does the error handling work?", + "context_provider_messages": test_messages, + } - result = node(test_state) + result = node(test_state) - assert "context_provider_messages" in result - assert len(result["context_provider_messages"]) == 1 - assert result["context_provider_messages"][0].content == fake_response + assert "context_provider_messages" in result + assert len(result["context_provider_messages"]) == 1 + assert result["context_provider_messages"][0].content == fake_response diff --git a/tests/neo4j/test_knowledge_graph_handler.py b/tests/neo4j/test_knowledge_graph_handler.py index adddd3d2..6d3888aa 100644 --- a/tests/neo4j/test_knowledge_graph_handler.py +++ b/tests/neo4j/test_knowledge_graph_handler.py @@ -11,109 +11,102 @@ @pytest.mark.slow async def test_num_ast_nodes(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_ast_nodes = session.execute_read(handler._read_ast_nodes, root_node_id=0) - assert len(read_ast_nodes) == 84 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_ast_nodes = session.execute_read(handler._read_ast_nodes, root_node_id=0) + assert len(read_ast_nodes) == 84 @pytest.mark.slow async def test_num_file_nodes(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_file_nodes = session.execute_read(handler._read_file_nodes, root_node_id=0) - assert len(read_file_nodes) == 9 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_file_nodes = session.execute_read(handler._read_file_nodes, root_node_id=0) + assert len(read_file_nodes) == 9 @pytest.mark.slow async def test_num_text_nodes(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_text_nodes = session.execute_read(handler._read_text_nodes, root_node_id=0) - assert len(read_text_nodes) == 2 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_text_nodes = session.execute_read(handler._read_text_nodes, root_node_id=0) + assert len(read_text_nodes) == 2 @pytest.mark.slow async def test_num_parent_of_edges(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_parent_of_edges = session.execute_read( - handler._read_parent_of_edges, root_node_id=0 - ) - assert len(read_parent_of_edges) == 81 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_parent_of_edges = session.execute_read( + handler._read_parent_of_edges, root_node_id=0 + ) + assert len(read_parent_of_edges) == 81 @pytest.mark.slow async def test_num_has_file_edges(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_has_file_edges = session.execute_read( - handler._read_has_file_edges, root_node_id=0 - ) - print(read_has_file_edges) - assert len(read_has_file_edges) == 8 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_has_file_edges = session.execute_read(handler._read_has_file_edges, root_node_id=0) + assert len(read_has_file_edges) == 8 @pytest.mark.slow async def test_num_has_ast_edges(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_has_ast_edges = session.execute_read( - handler._read_has_ast_edges, root_node_id=0 - ) - assert len(read_has_ast_edges) == 3 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_has_ast_edges = session.execute_read(handler._read_has_ast_edges, root_node_id=0) + assert len(read_has_ast_edges) == 3 @pytest.mark.slow async def test_num_has_text_edges(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_has_text_edges = session.execute_read( - handler._read_has_text_edges, root_node_id=0 - ) - assert len(read_has_text_edges) == 2 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_has_text_edges = session.execute_read(handler._read_has_text_edges, root_node_id=0) + assert len(read_has_text_edges) == 2 @pytest.mark.slow async def test_num_next_chunk_edges(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - with neo4j_container.get_driver() as driver: - with driver.session() as session: - read_next_chunk_edges = session.execute_read( - handler._read_next_chunk_edges, root_node_id=0 - ) - assert len(read_next_chunk_edges) == 1 + with neo4j_container.get_driver() as driver: + with driver.session() as session: + read_next_chunk_edges = session.execute_read( + handler._read_next_chunk_edges, root_node_id=0 + ) + assert len(read_next_chunk_edges) == 1 @pytest.mark.slow async def test_knowledge_graph_exists(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) + neo4j_container, kg = neo4j_container_with_kg_fixture + handler = KnowledgeGraphHandler(neo4j_container.get_driver(), 100) - assert handler.knowledge_graph_exists(0) + assert handler.knowledge_graph_exists(0) @pytest.mark.slow diff --git a/tests/tools/test_graph_traversal.py b/tests/tools/test_graph_traversal.py index 39723e7b..86ccbdf1 100644 --- a/tests/tools/test_graph_traversal.py +++ b/tests/tools/test_graph_traversal.py @@ -7,24 +7,24 @@ @pytest.mark.slow async def test_find_file_node_with_basename(neo4j_container_with_kg_fixture): # noqa: F811 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_file_node_with_basename( - test_project_paths.PYTHON_FILE.name, driver, 1000 - ) - - basename = test_project_paths.PYTHON_FILE.name - relative_path = str( - test_project_paths.PYTHON_FILE.relative_to( - test_project_paths.TEST_PROJECT_PATH - ).as_posix() - ) - - result_data = result[1] - assert len(result_data) == 1 - assert "FileNode" in result_data[0] - assert result_data[0]["FileNode"].get("basename", "") == basename - assert result_data[0]["FileNode"].get("relative_path", "") == relative_path + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_file_node_with_basename( + test_project_paths.PYTHON_FILE.name, driver, 1000 + ) + + basename = test_project_paths.PYTHON_FILE.name + relative_path = str( + test_project_paths.PYTHON_FILE.relative_to( + test_project_paths.TEST_PROJECT_PATH + ).as_posix() + ) + + result_data = result[1] + assert len(result_data) == 1 + assert "FileNode" in result_data[0] + assert result_data[0]["FileNode"].get("basename", "") == basename + assert result_data[0]["FileNode"].get("relative_path", "") == relative_path @pytest.mark.slow @@ -32,35 +32,35 @@ async def test_find_file_node_with_relative_path(neo4j_container_with_kg_fixture relative_path = str( test_project_paths.MD_FILE.relative_to(test_project_paths.TEST_PROJECT_PATH).as_posix() ) - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_file_node_with_relative_path(relative_path, driver, 1000) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_file_node_with_relative_path(relative_path, driver, 1000) - basename = test_project_paths.MD_FILE.name + basename = test_project_paths.MD_FILE.name - result_data = result[1] - assert len(result_data) == 1 - assert "FileNode" in result_data[0] - assert result_data[0]["FileNode"].get("basename", "") == basename - assert result_data[0]["FileNode"].get("relative_path", "") == relative_path + result_data = result[1] + assert len(result_data) == 1 + assert "FileNode" in result_data[0] + assert result_data[0]["FileNode"].get("basename", "") == basename + assert result_data[0]["FileNode"].get("relative_path", "") == relative_path @pytest.mark.slow async def test_find_ast_node_with_text_in_file_with_basename(neo4j_container_with_kg_fixture): # noqa: F811 basename = test_project_paths.PYTHON_FILE.name - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_ast_node_with_text_in_file_with_basename( - "Hello world!", basename, driver, 1000 - ) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_ast_node_with_text_in_file_with_basename( + "Hello world!", basename, driver, 1000 + ) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "ASTNode" in result_row - assert "Hello world!" in result_row["ASTNode"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("basename", "") == basename + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "ASTNode" in result_row + assert "Hello world!" in result_row["ASTNode"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("basename", "") == basename @pytest.mark.slow @@ -68,38 +68,38 @@ async def test_find_ast_node_with_text_in_file_with_relative_path(neo4j_containe relative_path = str( test_project_paths.C_FILE.relative_to(test_project_paths.TEST_PROJECT_PATH).as_posix() ) - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_ast_node_with_text_in_file_with_relative_path( - "Hello world!", relative_path, driver, 1000 - ) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_ast_node_with_text_in_file_with_relative_path( + "Hello world!", relative_path, driver, 1000 + ) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "ASTNode" in result_row - assert "Hello world!" in result_row["ASTNode"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == relative_path + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "ASTNode" in result_row + assert "Hello world!" in result_row["ASTNode"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == relative_path @pytest.mark.slow async def test_find_ast_node_with_type_in_file_with_basename(neo4j_container_with_kg_fixture): # noqa: F811 basename = test_project_paths.C_FILE.name node_type = "function_definition" - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_ast_node_with_type_in_file_with_basename( - node_type, basename, driver, 1000 - ) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_ast_node_with_type_in_file_with_basename( + node_type, basename, driver, 1000 + ) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "ASTNode" in result_row - assert result_row["ASTNode"].get("type", "") == node_type - assert "FileNode" in result_row - assert result_row["FileNode"].get("basename", "") == basename + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "ASTNode" in result_row + assert result_row["ASTNode"].get("type", "") == node_type + assert "FileNode" in result_row + assert result_row["FileNode"].get("basename", "") == basename @pytest.mark.slow @@ -108,97 +108,96 @@ async def test_find_ast_node_with_type_in_file_with_relative_path(neo4j_containe test_project_paths.JAVA_FILE.relative_to(test_project_paths.TEST_PROJECT_PATH).as_posix() ) node_type = "string_literal" - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_ast_node_with_type_in_file_with_relative_path( - node_type, relative_path, driver, 1000 - ) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_ast_node_with_type_in_file_with_relative_path( + node_type, relative_path, driver, 1000 + ) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "ASTNode" in result_row - assert result_row["ASTNode"].get("type", "") == node_type - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == relative_path + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "ASTNode" in result_row + assert result_row["ASTNode"].get("type", "") == node_type + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == relative_path @pytest.mark.slow async def test_find_text_node_with_text(neo4j_container_with_kg_fixture): # noqa: F811 text = "Text under header C" - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_text_node_with_text(text, driver, 1000) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_text_node_with_text(text, driver, 1000) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "TextNode" in result_row - assert text in result_row["TextNode"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == "foo/test.md" + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "TextNode" in result_row + assert text in result_row["TextNode"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == "foo/test.md" @pytest.mark.slow async def test_find_text_node_with_text_in_file(neo4j_container_with_kg_fixture): # noqa: F811 basename = test_project_paths.MD_FILE.name text = "Text under header B" - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.find_text_node_with_text_in_file(text, basename, driver, 1000) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.find_text_node_with_text_in_file(text, basename, driver, 1000) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "TextNode" in result_row - assert text in result_row["TextNode"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("basename", "") == basename + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "TextNode" in result_row + assert text in result_row["TextNode"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("basename", "") == basename @pytest.mark.slow async def test_get_next_text_node_with_node_id(neo4j_container_with_kg_fixture): # noqa: F811 node_id = 36 - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.get_next_text_node_with_node_id(node_id, driver, 1000) + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.get_next_text_node_with_node_id(node_id, driver, 1000) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "TextNode" in result_row - assert "Text under header D" in result_row["TextNode"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == "foo/test.md" + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "TextNode" in result_row + assert "Text under header D" in result_row["TextNode"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == "foo/test.md" @pytest.mark.slow async def test_preview_file_content_with_basename(neo4j_container_with_kg_fixture): # noqa: F811 basename = test_project_paths.PYTHON_FILE.name - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.preview_file_content_with_basename(basename, driver, 1000) - - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "preview" in result_row - assert 'print("Hello world!")' in result_row["preview"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("basename", "") == basename + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.preview_file_content_with_basename(basename, driver, 1000) + + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "preview" in result_row + assert 'print("Hello world!")' in result_row["preview"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("basename", "") == basename basename = test_project_paths.MD_FILE.name - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.preview_file_content_with_basename(basename, driver, 1000) + with neo4j_container.get_driver() as driver: + result = graph_traversal.preview_file_content_with_basename(basename, driver, 1000) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "preview" in result_row - assert "Text under header A" in result_row["preview"] - assert "FileNode" in result_row - assert result_row["FileNode"].get("basename", "") == basename + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "preview" in result_row + assert "Text under header A" in result_row["preview"] + assert "FileNode" in result_row + assert result_row["FileNode"].get("basename", "") == basename @pytest.mark.slow @@ -206,54 +205,53 @@ async def test_preview_file_content_with_relative_path(neo4j_container_with_kg_f relative_path = str( test_project_paths.PYTHON_FILE.relative_to(test_project_paths.TEST_PROJECT_PATH).as_posix() ) - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.preview_file_content_with_relative_path( - relative_path, driver, 1000 - ) - - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "preview" in result_row - assert 'print("Hello world!")' in result_row["preview"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == relative_path + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.preview_file_content_with_relative_path( + relative_path, driver, 1000 + ) + + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "preview" in result_row + assert 'print("Hello world!")' in result_row["preview"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == relative_path relative_path = str( test_project_paths.MD_FILE.relative_to(test_project_paths.TEST_PROJECT_PATH).as_posix() ) - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.preview_file_content_with_relative_path( - relative_path, driver, 1000 - ) + with neo4j_container.get_driver() as driver: + result = graph_traversal.preview_file_content_with_relative_path( + relative_path, driver, 1000 + ) - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "preview" in result_row - assert "Text under header A" in result_row["preview"] - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == relative_path + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "preview" in result_row + assert "Text under header A" in result_row["preview"] + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == relative_path @pytest.mark.slow async def test_read_code_with_basename(neo4j_container_with_kg_fixture): # noqa: F811 basename = test_project_paths.JAVA_FILE.name - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.read_code_with_basename(basename, 2, 3, driver, 1000) - - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "SelectedLines" in result_row - assert "public static void main(String[] args) {" in result_row[ - "SelectedLines" - ].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("basename", "") == basename + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.read_code_with_basename(basename, 2, 3, driver, 1000) + + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "SelectedLines" in result_row + assert "public static void main(String[] args) {" in result_row["SelectedLines"].get( + "text", "" + ) + assert "FileNode" in result_row + assert result_row["FileNode"].get("basename", "") == basename @pytest.mark.slow @@ -261,14 +259,14 @@ async def test_read_code_with_relative_path(neo4j_container_with_kg_fixture): # relative_path = str( test_project_paths.C_FILE.relative_to(test_project_paths.TEST_PROJECT_PATH).as_posix() ) - async for neo4j_container, kg in neo4j_container_with_kg_fixture: - with neo4j_container.get_driver() as driver: - result = graph_traversal.read_code_with_relative_path(relative_path, 5, 6, driver, 1000) - - result_data = result[1] - assert len(result_data) > 0 - for result_row in result_data: - assert "SelectedLines" in result_row - assert "return 0;" in result_row["SelectedLines"].get("text", "") - assert "FileNode" in result_row - assert result_row["FileNode"].get("relative_path", "") == relative_path + neo4j_container, kg = neo4j_container_with_kg_fixture + with neo4j_container.get_driver() as driver: + result = graph_traversal.read_code_with_relative_path(relative_path, 5, 6, driver, 1000) + + result_data = result[1] + assert len(result_data) > 0 + for result_row in result_data: + assert "SelectedLines" in result_row + assert "return 0;" in result_row["SelectedLines"].get("text", "") + assert "FileNode" in result_row + assert result_row["FileNode"].get("relative_path", "") == relative_path