test: test for job

lcserve/backend/gateway.py

@@ -670,7 +670,9 @@ async def job(
                         + '-'
                         + os.getenv('K8S_NAMESPACE_NAME').split('-')[1]
                     )
-                    job_name = func.__name__ + '-' + uuid.uuid4().hex[:5]
+                    job_name = (
+                        func.__name__.replace('_', '-') + '-' + uuid.uuid4().hex[:5]
+                    )
                     image_id = os.getenv('LCSERVE_IMAGE')
                     entrypoint = [
                         'python',

tests/integration/apps/basic_app.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import time
 from typing import Dict, List
 
@@ -9,7 +10,7 @@
 from langchain.callbacks.manager import CallbackManager
 from langchain.llms.fake import FakeListLLM
 
-from lcserve import serving
+from lcserve import job, serving
 
 
 @serving
@@ -198,3 +199,8 @@ def tracing_ws(dummy: str, **kwargs):
     agent.run(dummy)
 
     return 'ok'
+
+
+@job(timeout=100, backofflimit=3)
+def my_job(param1: str, param2: str):
+    print('ok')

tests/integration/jcloud/test_basic_app.py

@@ -1,9 +1,11 @@
+import asyncio
 import json
 import time
 
 import pytest
 import requests
 import websockets
+from jcloud.flow import CloudFlow
 from websockets.exceptions import ConnectionClosedOK
 
 from ..helper import deploy_jcloud_app
@@ -17,6 +19,7 @@ async def test_basic_app():
         await _test_workspace(app_id)
         await _test_local_file_access(app_id)
         await _test_tracing(app_id)
+        await _test_job(app_id)
 
 
 def _test_http_route(app_id):
@@ -107,3 +110,22 @@ async def _test_tracing(app_id: str):
         assert "dummy string ws" in message
     except ConnectionClosedOK:
         pass
+
+
+async def _test_job(app_id: str):
+    asyncio.sleep(30)
+    create_job_url = f"https://{app_id}.wolf.jina.ai/my_job"
+
+    response = requests.post(
+        create_job_url, data=json.dumps({"param1": "hello", "param2": "world"})
+    )
+    assert response.status_code == 200
+
+    asyncio.sleep(30)
+
+    jobs = await CloudFlow(flow_id=app_id).list_resources("jobs")
+    assert (
+        len(jobs) > 1
+        and jobs[0]['name'].startswith('my-job-')
+        and jobs[0]['status']['conditions'][-1]['type'] == 'Complete'
+    )