feat(cli.py): add json_output option to datalines_list command for JSON format output

fix(cli.py): handle project selection error more gracefully in datalines_select command
test(cli.py): add tests for datalines_list and datalines_get commands to ensure functionality and output correctness

Author: closedLoop

infactory_client/cli.py

@@ -612,7 +612,10 @@ def datasource_create(
 
 @datalines_app.command(name="list")
 def datalines_list(
-    project_id: str | None = typer.Option(None, help="Project ID to list datalines for")
+    project_id: str | None = typer.Option(
+        None, help="Project ID to list datalines for"
+    ),
+    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
 ):
     """List datalines."""
     client = get_client()
@@ -632,13 +635,32 @@ def datalines_list(
             typer.echo("No datalines found")
             return
 
+        if json_output:
+            datalines_data = [
+                {
+                    "id": dl.id,
+                    "name": dl.name
+                    or dl.data_model.get("schemaT", {}).get("name", dl.name),
+                    # "data_model": dl.data_model,
+                    # "schema_code": dl.schema_code,
+                    "project_id": dl.project_id,
+                    "dataobject_id": dl.dataobject_id,
+                    # "created_at": dl.created_at,
+                    # "updated_at": dl.updated_at,
+                    # "deleted_at": dl.deleted_at,
+                }
+                for dl in datalines
+            ]
+            print(json.dumps(datalines_data))
+            return
+
         table = Table()
         table.add_column("ID")
         table.add_column("Name")
 
         for dl in datalines:
             name = dl.name
-            if dl.data_model.get("schemaT"):
+            if dl.data_model and dl.data_model.get("schemaT"):
                 name = dl.data_model["schemaT"].get("name", name) or name
             table.add_row(dl.id, name)
 
@@ -660,16 +682,15 @@ def datalines_select(
     if model and code:
         typer.echo("Error: --model and --code cannot be used together", err=True)
         raise typer.Exit(1)
-    if not model and not code:
-        model = True
 
-    try:
-        if not client.state.project_id:
-            typer.echo(
-                "No project selected. Please select a project first with 'nf projects select'"
-            )
-            return
+    if not client.state.project_id:
+        typer.echo(
+            "No project selected. Please select a project first with 'nf projects select'",
+            err=True,
+        )
+        raise typer.Exit(1)
 
+    try:
         datalines = client.datalines.list(project_id=client.state.project_id)
 
         if not datalines:
@@ -686,10 +707,7 @@ def datalines_select(
         table.add_column("Name")
 
         for num, dl in choices.items():
-            name = dl.name
-            if dl.data_model.get("schemaT"):
-                name = dl.data_model["schemaT"].get("name", name) or name
-            table.add_row(num, dl.id, name)
+            table.add_row(num, dl.id, dl.name)
 
         console.print(table)
 
@@ -703,12 +721,12 @@ def datalines_select(
         selected_dataline = choices[choice]
 
         if model and selected_dataline.data_model:
-            typer.echo(json.dumps(selected_dataline.data_model, indent=2))
+            print(json.dumps(selected_dataline.data_model, indent=2))
         elif code and selected_dataline.schema_code:
-            typer.echo(selected_dataline.schema_code)
+            print(selected_dataline.schema_code)
         else:
             typer.echo(
-                f"Dataline: {selected_dataline.name} (ID: {selected_dataline.id})"
+                f"\nSelected dataline: {selected_dataline.name} (ID: {selected_dataline.id})"
             )
 
     except Exception as e:
@@ -725,13 +743,11 @@ def datalines_get(
     """Get a dataline by ID."""
     client = get_client()
 
-    if model and code:
-        typer.echo("Error: --model and --code cannot be used together", err=True)
-        raise typer.Exit(1)
-    if not model and not code:
-        model = True
-
     try:
+        if model and code:
+            typer.echo("Error: --model and --code cannot be used together", err=True)
+            raise typer.Exit(1)
+
         dataline = client.datalines.get(dataline_id)
 
         if model and dataline.data_model:

tests/test_cli_resources.py

@@ -398,6 +398,217 @@ def test_datasource_create(self):
         self.assertIn("Type: mysql", result.stdout)
         self.assertIn("URI:", result.stdout)
 
+    def test_datalines_list(self):
+        """Test dataline listing"""
+        # First select an organization, team, and project
+        result = self.runner.invoke(app, ["orgs", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        orgs_data = json.loads(result.stdout)
+
+        if not orgs_data:
+            self.skipTest("No organizations available for testing")
+
+        # Set organization
+        org_id = orgs_data[0]["id"]
+        result = self.runner.invoke(app, ["orgs", "set", org_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Get and set team
+        result = self.runner.invoke(app, ["teams", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        teams_data = json.loads(result.stdout)
+
+        if not teams_data:
+            self.skipTest("No teams available for testing")
+
+        team_id = teams_data[0]["id"]
+        result = self.runner.invoke(app, ["teams", "set", team_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Get and set project
+        result = self.runner.invoke(app, ["projects", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        projects_data = json.loads(result.stdout)
+
+        if not projects_data:
+            self.skipTest("No projects available for testing")
+
+        project_id = projects_data[0]["id"]
+        result = self.runner.invoke(app, ["projects", "set", project_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Test datalines list without project ID (using current project)
+        result = self.runner.invoke(app, ["datalines", "list"])
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn("ID", result.stdout)
+        self.assertIn("Name", result.stdout)
+
+        # Test datalines list with explicit project ID
+        result = self.runner.invoke(
+            app, ["datalines", "list", "--project-id", project_id]
+        )
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn("ID", result.stdout)
+        self.assertIn("Name", result.stdout)
+
+        # Test JSON output
+        result = self.runner.invoke(app, ["datalines", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        if "No datalines found" not in result.stdout:
+            datalines_data = json.loads(result.stdout)
+            self.assertIsInstance(datalines_data, list)
+            if datalines_data:  # If there are datalines
+                self.assertIn("id", datalines_data[0])
+                self.assertIn("name", datalines_data[0])
+
+        # Test JSON output with explicit project ID
+        result = self.runner.invoke(
+            app, ["datalines", "list", "--json", "--project-id", project_id]
+        )
+        self.assertEqual(result.exit_code, 0)
+        if "No datalines found" not in result.stdout:
+            datalines_data = json.loads(result.stdout)
+            self.assertIsInstance(datalines_data, list)
+            if datalines_data:  # If there are datalines
+                self.assertIn("id", datalines_data[0])
+                self.assertIn("name", datalines_data[0])
+
+    def test_datalines_get(self):
+        """Test getting a dataline by ID"""
+        # First select an organization, team, and project
+        result = self.runner.invoke(app, ["orgs", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        orgs_data = json.loads(result.stdout)
+
+        if not orgs_data:
+            self.skipTest("No organizations available for testing")
+
+        # Set organization
+        org_id = orgs_data[0]["id"]
+        result = self.runner.invoke(app, ["orgs", "set", org_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Get and set team
+        result = self.runner.invoke(app, ["teams", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        teams_data = json.loads(result.stdout)
+
+        if not teams_data:
+            self.skipTest("No teams available for testing")
+
+        team_id = teams_data[0]["id"]
+        result = self.runner.invoke(app, ["teams", "set", team_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Get and set project
+        result = self.runner.invoke(app, ["projects", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        projects_data = json.loads(result.stdout)
+
+        if not projects_data:
+            self.skipTest("No projects available for testing")
+
+        project_id = projects_data[0]["id"]
+        result = self.runner.invoke(app, ["projects", "set", project_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # First get a list of datalines
+        result = self.runner.invoke(app, ["datalines", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+
+        datalines = json.loads(result.stdout)
+
+        # If we have datalines, test get command
+        if len(datalines) > 0:
+            # Get the first dataline ID from the output
+            dataline_id = datalines[0]["id"]
+
+            # Test basic get
+            result = self.runner.invoke(app, ["datalines", "get", dataline_id])
+            self.assertEqual(result.exit_code, 0)
+            self.assertIn(dataline_id, result.stdout)
+
+            # Test get with --model flag
+            result = self.runner.invoke(
+                app, ["datalines", "get", dataline_id, "--model"]
+            )
+            self.assertEqual(result.exit_code, 0)
+            # Note: We can't guarantee valid JSON output as it depends on the data
+
+            # Test get with --code flag
+            result = self.runner.invoke(
+                app, ["datalines", "get", dataline_id, "--code"]
+            )
+            self.assertEqual(result.exit_code, 0)
+            # Note: We can't guarantee schema code exists
+
+            # Test error case: model and code flags together
+            result = self.runner.invoke(
+                app, ["datalines", "get", dataline_id, "--model", "--code"]
+            )
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(
+                "Error: --model and --code cannot be used together", result.stdout
+            )
+
+            # Test error case: invalid ID
+            result = self.runner.invoke(app, ["datalines", "get", "invalid-id"])
+            self.assertNotEqual(result.exit_code, 0)
+        else:
+            self.skipTest("No datalines available for testing")
+
+    def test_datalines_select(self):
+        """Test interactive dataline selection"""
+        # First select an organization, team, and project
+        result = self.runner.invoke(app, ["orgs", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        orgs_data = json.loads(result.stdout)
+
+        if not orgs_data:
+            self.skipTest("No organizations available for testing")
+
+        # Set organization
+        org_id = orgs_data[0]["id"]
+        result = self.runner.invoke(app, ["orgs", "set", org_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Get and set team
+        result = self.runner.invoke(app, ["teams", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        teams_data = json.loads(result.stdout)
+
+        if not teams_data:
+            self.skipTest("No teams available for testing")
+
+        team_id = teams_data[0]["id"]
+        result = self.runner.invoke(app, ["teams", "set", team_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Get and set project
+        result = self.runner.invoke(app, ["projects", "list", "--json"])
+        self.assertEqual(result.exit_code, 0)
+        projects_data = json.loads(result.stdout)
+
+        if not projects_data:
+            self.skipTest("No projects available for testing")
+
+        project_id = projects_data[0]["id"]
+        result = self.runner.invoke(app, ["projects", "set", project_id])
+        self.assertEqual(result.exit_code, 0)
+
+        # Test error case: model and code flags together
+        result = self.runner.invoke(app, ["datalines", "select", "--model", "--code"])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(
+            "Error: --model and --code cannot be used together", result.stdout
+        )
+
+        # Test basic select command (will show prompt but not proceed due to no input)
+        result = self.runner.invoke(app, ["datalines", "select"])
+        if "No datalines found" in result.stdout:
+            self.skipTest("No datalines available for testing")
+        self.assertIn("Select dataline number", result.stdout)
+
 
 if __name__ == "__main__":
     unittest.main()