1
# Check that multiprocessing top level import works
2
import multiprocessing
3
 
4
from workers import WorkerEntrypoint, import_from_javascript
5
 
6
from pyodide import __version__
7
 
8
 
9
def use(x):
10
    pass
11
 
12
 
13
def f():
14
    pass
15
 
16
 
17
def top_level_test():
18
    # Test the regular import function
19
    workers_module = import_from_javascript("cloudflare:workers")
20
    env = workers_module.env
21
    assert env.FOO == "BAR", "env.FOO should be 'BAR'"
22
    sockets_module = import_from_javascript("cloudflare:sockets")
23
 
24
    vectorize = None
25
    threw = False
26
    try:
27
        vectorize = import_from_javascript("cloudflare:vectorize")
28
    except ImportError:
29
        threw = True
30
 
31
    if __version__ == "0.26.0a2":
32
        # Should have thrown throw "ImportError: Failed to import 'cloudflare:vectorize': Only 'cloudflare:workers' and 'cloudflare:sockets' are available until the next python runtime version."
33
        msg = 'import_from_javascript("cloudflare:vectorize") not expected to work in the global scope in 0.26.0a2'
34
        assert threw, msg
35
        assert vectorize is None, msg
36
    else:
37
        assert {"DistanceMetric", "KnownModel", "MetadataRetrievalLevel"}.issubset(
38
            dir(vectorize)
39
        )
40
 
41
    from js import setTimeout
42
 
43
    from pyodide.ffi import create_once_callable
44
 
45
    # Checks for the patch to top level setTimeout() that is needed for top level
46
    # `import_from_javascript``.
47
    # 1. Test that setTimeout with a zero timeout works at top level
48
    setTimeout(create_once_callable(f), 0)
49
 
50
    # 2. Test that setTimeout with nonzero timeout fails with the expected error.
51
    err = None
52
    try:
53
        setTimeout(create_once_callable(f), 1)
54
    except Exception as e:
55
        err = e
56
    finally:
57
        assert err
58
        assert str(err).startswith(
59
            "Error: Disallowed operation called within global scope"
60
        )
61
        del err
62
    return [workers_module, env, sockets_module, vectorize]
63
 
64
 
65
# Check serialization of modules
66
TOP_LEVEL = top_level_test()
67
 
68
 
69
class Default(WorkerEntrypoint):
70
    async def test(self):
71
        use(multiprocessing)