added tests

tests/sources/rest_api/conftest.py

@@ -130,6 +130,18 @@ def post_comments(request, context):
             post_id = int(request.url.split("/")[-2])
             return paginate_by_page_number(request, generate_comments(post_id))
 
+        @router.get(r"/posts/(\d+)/protected_comments")
+        def protected_basic_auth(request, context):
+            post_id = int(request.url.split("/")[-2])
+            post_id_from_header = request.headers.get("Post_id")
+            auth = request.headers.get("Authorization")
+            creds = "user:password"
+            creds_base64 = base64.b64encode(creds.encode()).decode()
+            if auth == f"Basic {creds_base64}" and post_id == post_id_from_header:
+                return paginate_by_page_number(request, generate_comments(post_id))
+            context.status_code = 401
+            return {"error": "Unauthorized"}
+
         @router.get(r"/posts/\d+$")
         def post_detail(request, context):
             post_id = request.url.split("/")[-1]

tests/sources/rest_api/integration/test_offline.py

@@ -168,6 +168,70 @@ def test_load_mock_api_with_json_resolved(mock_api_server):
         full_refresh=True,
     )
 
+    mock_source = rest_api_source(
+        {
+            "client": {"base_url": "https://api.example.com"},
+            "resources": [
+                "posts",
+                {
+                    "name": "post_details",
+                    "endpoint": {
+                        "path": "posts/search_by_id",
+                        "method": "POST",
+                        "json": {
+                            "post_id": "{posts_id}",
+                        },
+                        "params": {
+                            "posts_id": {
+                                "type": "resolve",
+                                "resource": "posts",
+                                "field": "id",
+                            }
+                        },
+                    },
+                },
+            ],
+        }
+    )
+
+    load_info = pipeline.run(mock_source)
+    print(load_info)
+    assert_load_info(load_info)
+    table_names = [t["name"] for t in pipeline.default_schema.data_tables()]
+    table_counts = load_table_counts(pipeline, *table_names)
+
+    assert table_counts.keys() == {"posts", "post_details"}
+
+    assert table_counts["posts"] == DEFAULT_PAGE_SIZE * DEFAULT_TOTAL_PAGES
+    assert table_counts["post_details"] == DEFAULT_PAGE_SIZE * DEFAULT_TOTAL_PAGES
+
+    with pipeline.sql_client() as client:
+        posts_table = client.make_qualified_table_name("posts")
+        posts_details_table = client.make_qualified_table_name("post_details")
+
+    print(pipeline.default_schema.to_pretty_yaml())
+
+    assert_query_data(
+        pipeline,
+        f"SELECT title FROM {posts_table} ORDER BY id limit 25",
+        [f"Post {i}" for i in range(25)],
+    )
+
+    assert_query_data(
+        pipeline,
+        f"SELECT body FROM {posts_details_table} ORDER BY id limit 25",
+        [f"Post body {i}" for i in range(25)],
+    )
+
+
+def test_load_mock_api_with_json_resolved_with_implicit_param(mock_api_server):
+    pipeline = dlt.pipeline(
+        pipeline_name="rest_api_mock",
+        destination="duckdb",
+        dataset_name="rest_api_mock",
+        full_refresh=True,
+    )
+
     mock_source = rest_api_source(
         {
             "client": {"base_url": "https://api.example.com"},