Skip to content

updated tests for flows endpoint #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ag-nexla wants to merge 1 commit into
base: main
Choose a base branch
from
Open

updated tests for flows endpoint #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 96 additions & 45 deletions tests/integration/test_flows.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,28 +25,55 @@ def client(self) -> Optional[NexlaClient]:

return NexlaClient(**creds)

@pytest.fixture(scope="class")
def test_flow_id(self) -> Optional[int]:
"""Get test flow ID from environment."""
@pytest.fixture(scope="function")
def test_flow_id(self, client) -> Optional[int]:
"""Get test flow ID dynamically or from environment."""
# Try environment variable first
flow_id = os.getenv("TEST_FLOW_ID")
if flow_id:
return int(flow_id)

# Fallback: get first available flow
try:
flows = client.flows.list()
if flows and flows[0].flows:
return flows[0].flows[0].id
except Exception:
pass
return None

@pytest.fixture(scope="class")
def test_source_id(self) -> Optional[int]:
"""Get test source ID from environment."""
@pytest.fixture(scope="function")
def test_source_id(self, client) -> Optional[int]:
"""Get test source ID dynamically or from environment."""
# Try environment variable first
source_id = os.getenv("TEST_SOURCE_ID")
if source_id:
return int(source_id)

# Fallback: get first available source
try:
sources = client.sources.list()
if sources:
return sources[0].id
except Exception:
pass
return None

@pytest.fixture(scope="class")
def test_dataset_id(self) -> Optional[int]:
"""Get test dataset ID from environment."""
@pytest.fixture(scope="function")
def test_dataset_id(self, client) -> Optional[int]:
"""Get test dataset ID dynamically or from environment."""
# Try environment variable first
dataset_id = os.getenv("TEST_DATASET_ID")
if dataset_id:
return int(dataset_id)

# Fallback: get first available dataset
try:
datasets = client.nexsets.list()
if datasets:
return datasets[0].id
except Exception:
pass
return None

def test_list_flows(self, client):
Expand Down Expand Up @@ -154,21 +181,31 @@ def test_get_flow_by_dataset(self, client, test_dataset_id):
if not test_dataset_id:
pytest.skip("No test dataset ID provided")

# Act
flow = client.flows.get_by_resource("data_sets", test_dataset_id)

# Assert
assert isinstance(flow, FlowResponse)
assert isinstance(flow.flows, list)

# Verify the flow contains the dataset
if flow.flows:
# Nodes should have data_set_id matching
dataset_found = any(
getattr(node, 'data_set_id', None) == test_dataset_id
for node in flow.flows
)
assert dataset_found, f"Dataset ID {test_dataset_id} not found in flow"
try:
# Act
flow = client.flows.get_by_resource("data_sets", test_dataset_id)

# Assert
assert isinstance(flow, FlowResponse)
assert isinstance(flow.flows, list)

# Verify the flow contains the dataset
if flow.flows:
# Nodes should have data_set_id matching
dataset_found = any(
getattr(node, 'data_set_id', None) == test_dataset_id
for node in flow.flows
)
assert dataset_found, f"Dataset ID {test_dataset_id} not found in flow"
else:
# If no flows returned, that's also valid - dataset might not be in any flow
pytest.skip("No flows found for this dataset")

except Exception as e:
# If the API call fails (e.g., dataset not in any flow), skip the test
if "not found" in str(e).lower() or "404" in str(e):
pytest.skip(f"Dataset {test_dataset_id} not associated with any flow: {e}")
raise

def test_flow_activation_pause_cycle(self, client, test_flow_id):
"""Test activating and pausing a flow."""
Expand Down Expand Up @@ -246,6 +283,9 @@ def test_flow_copy(self, client, test_flow_id):
try:
# Pause before deleting
client.flows.pause(node.id, all=True)
# Wait 5 seconds for pause to take effect
import time
time.sleep(5)
client.flows.delete(node.id)
except ServerError:
pass # Best effort cleanup
Expand All @@ -256,32 +296,43 @@ def test_flow_copy(self, client, test_flow_id):
raise

def test_delete_flow_validation(self, client):
"""Test that deleting active flow fails with proper error."""
"""Test that deleting non-existent flow fails with proper error."""
# We don't actually want to delete real flows in integration tests
# Just verify the error handling works
# Just verify the error handling works for non-existent flows

# Find an active flow
flows = client.flows.list()
active_flow_id = None
# Use a non-existent flow ID
non_existent_flow_id = 999999999

for flow_resp in flows:
for node in flow_resp.flows:
# Assuming we can check status somehow
# This is a simplified test
active_flow_id = node.id
break
if active_flow_id:
break

if not active_flow_id:
pytest.skip("No active flow found for testing")
# Try to delete non-existent flow - should fail
from nexla_sdk.exceptions import NotFoundError
with pytest.raises(NotFoundError):
client.flows.delete(non_existent_flow_id)

def test_delete_active_flow_validation(self, client, test_flow_id):
"""Test that deleting active flow is prevented (either by error or silent ignore)."""
if not test_flow_id:
pytest.skip("No test flow ID provided")

# Try to delete active flow - should fail
with pytest.raises(ServerError) as exc_info:
client.flows.delete(active_flow_id)
# Get flow before deletion attempt
flow_before = client.flows.get(test_flow_id)
assert isinstance(flow_before, FlowResponse)

# Verify error is about active resources
assert exc_info.value.status_code in (400, 405)
try:
# Try to delete active flow - should either fail with error or be silently ignored
result = client.flows.delete(test_flow_id)

# If no exception was raised, verify the flow still exists (silent ignore)
flow_after = client.flows.get(test_flow_id)
assert isinstance(flow_after, FlowResponse)
assert flow_after.flows[0].id == test_flow_id, "Active flow deletion was not prevented"

except Exception as e:
# If an exception was raised, verify it's about active flow deletion
error_message = str(e).lower()
assert any(keyword in error_message for keyword in [
'active', 'running', 'cannot delete', 'delete not allowed',
'flow is active', 'must be paused', 'status'
]), f"Expected error about active flow, got: {e}"

def test_flow_structure_validation(self, client):
"""Test that flow structures are properly formed."""
Expand Down
Loading