FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
replicate-python/tests/test_deployment.py at main · hatgit/replicate-python · GitHub
hatgit
/
replicate-python
Public
forked from
replicate/replicate-python
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
replicate-python
/
tests
/
test_deployment.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
73 lines (63 loc) · 2.23 KB
Breadcrumbs
replicate-python
/
tests
/
test_deployment.py
Copy path
File metadata and controls
73 lines (63 loc) · 2.23 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import
json
import
httpx
import
pytest
import
respx
from
replicate
.
client
import
Client
router
=
respx
.
Router
(
base_url
=
"https://api.replicate.com/v1"
)
router
.
route
(
method
=
"POST"
,
path
=
"/deployments/test/model/predictions"
,
name
=
"deployments.predictions.create"
,
).
mock
(
return_value
=
httpx
.
Response
(
201
,
json
=
{
"id"
:
"p1"
,
"model"
:
"test/model"
,
"version"
:
"v1"
,
"urls"
: {
"get"
:
"https://api.replicate.com/v1/predictions/p1"
,
"cancel"
:
"https://api.replicate.com/v1/predictions/p1/cancel"
,
},
"created_at"
:
"2022-04-26T20:00:40.658234Z"
,
"source"
:
"api"
,
"status"
:
"processing"
,
"input"
: {
"text"
:
"world"
},
"output"
:
None
,
"error"
:
None
,
"logs"
:
""
,
},
)
)
router
.
route
(
host
=
"api.replicate.com"
).
pass_through
()
@
pytest
.
mark
.
asyncio
@
pytest
.
mark
.
parametrize
(
"async_flag"
, [
True
,
False
])
async
def
test_deployment_predictions_create
(
async_flag
):
client
=
Client
(
api_token
=
"test-token"
,
transport
=
httpx
.
MockTransport
(
router
.
handler
)
)
if
async_flag
:
deployment
=
await
client
.
deployments
.
async_get
(
"test/model"
)
prediction
=
await
deployment
.
predictions
.
async_create
(
input
=
{
"text"
:
"world"
},
webhook
=
"https://example.com/webhook"
,
webhook_events_filter
=
[
"completed"
],
stream
=
True
,
)
else
:
deployment
=
client
.
deployments
.
get
(
"test/model"
)
prediction
=
deployment
.
predictions
.
create
(
input
=
{
"text"
:
"world"
},
webhook
=
"https://example.com/webhook"
,
webhook_events_filter
=
[
"completed"
],
stream
=
True
,
)
assert
router
[
"deployments.predictions.create"
].
called
request
=
router
[
"deployments.predictions.create"
].
calls
[
0
].
request
request_body
=
json
.
loads
(
request
.
content
)
assert
request_body
[
"input"
]
==
{
"text"
:
"world"
}
assert
request_body
[
"webhook"
]
==
"https://example.com/webhook"
assert
request_body
[
"webhook_events_filter"
]
==
[
"completed"
]
assert
request_body
[
"stream"
]
is
True
assert
prediction
.
id
==
"p1"
assert
prediction
.
input
==
{
"text"
:
"world"
}
Back
|
FazBrowse Home
|
New Git URL