#!/usr/bin/env qore
# -*- mode: qore; indent-tabs-mode: nil -*-
%modern
%no-child-restrictions
%requires python
%requires QUnit
%requires reflection
%requires FsUtil
%try-module json
%define NO_JSON
%endtry
%requires WebSocketClient
# JNI is loaded dynamically at runtime in javaTest() to avoid conflicts with
# Python free-threading mode (PEP 703) - mimalloc is not compatible with JNI threads
%try-module xml
%define NO_XML
%endtry
%exec-class PythonTest
class PythonTest inherits Test {
public {
const ExprTests = (
{"expr": "1 + 1", "val": 2},
{"expr": "1.1", "val": 1.1},
{"expr": "True", "val": True},
{"expr": "False", "val": False},
{"expr": "'str'", "val": "str"},
{"expr": "bytes.fromhex('abcd')", "val": },
{"expr": "bytearray.fromhex('abcd')", "val": },
{"expr": "{'a': 1}", "val": {"a": 1}},
{"expr": "(1).bit_length()", "val": 1},
);
const Values = (
1,
1.1,
True,
False,
"str",
,
{"a": 1},
(1, 2, 3),
);
}
constructor() : Test("python test", "1.0") {
addTestCase("python thread test", \pythonThreadTest());
addTestCase("super test", \superTest());
addTestCase("java test", \javaTest());
addTestCase("exception test", \exceptionTest());
addTestCase("qore test", \qoreTest());
addTestCase("stack test", \stackTest());
addTestCase("thread test", \threadTest());
addTestCase("python callbacks", \pythonCallbackTest());
addTestCase("python import", \pythonImportTest());
addTestCase("object lifecycle", \objectLifecycleTest());
addTestCase("basic test", \basicTest());
addTestCase("named argument calls", \namedArgumentTest());
addTestCase("import test", \importTest());
addTestCase("qoreloader shared namespace import", \qoreloaderSharedNamespaceImportTest());
addTestCase("qoreloader multi-namespace import", \qoreloaderMultiNamespaceImportTest());
addTestCase("type conversion test", \typeConversionTest());
addTestCase("sandbox interrupt test", \sandboxInterruptTest());
# Set return value for compatibility with test harnesses that check the return value
set_return_value(main());
}
pythonThreadTest() {
# Python threads in sub-interpreters are not compatible with free-threading mode
# due to mimalloc heap initialization issues
if (Python::FreeThreading) {
testSkip("Python threads in sub-interpreters not supported with free-threading (PEP 703)");
}
{
PythonProgram p("
from threading import Thread
import qoreloader
from qore.__root__.Qore.Thread import Queue
def do_test(queue: Queue) -> None:
queue.push(1)
def test():
queue = Queue()
thread = Thread(target = do_test, args = (queue,))
thread.start();
thread.join()
return queue.get()
", "test.py");
auto o = p.callFunction("test");
assertEq(1, o);
}
{
PythonProgram p("
from threading import Thread
import qoreloader
from qore.__root__.Qore.Thread import Queue
class PyQueue(Queue):
def py_push(self, x):
self.push(x)
def do_test(queue: PyQueue):
queue.py_push(1)
def test():
queue = PyQueue()
thread = Thread(target = do_test, args = (queue,))
thread.start();
thread.join()
return queue.get()
", "test.py");
auto o = p.callFunction("test");
assertEq(1, o);
}
{
Program p(PO_NEW_STYLE);
p.importClass("QTest");
p.issueModuleCmd("python", "import-ns QTest x");
p.issueModuleCmd("python", "parse test from threading import Thread
import qoreloader
from qore.__root__.Qore.Thread import Counter
class Other:
def __init__(self) -> None:
self.py_test = PyTest()
def test(self, x) -> None:
self.py_test.process(x)
def do_test(self, counter: Counter) -> None:
self.test(1)
counter.dec()
class PyTest(x.QTest):
def test(self, x) -> None:
self.process(x)
def get_other() -> Other:
return Other()
def py_test(other: Other) -> int:
counter = Counter(1)
thread = Thread(target = other.do_test, args = (counter,))
thread.start();
counter.waitForZero()
thread.join()
return 1
");
p.issueModuleCmd("python", "export-func get_other");
p.issueModuleCmd("python", "export-func py_test");
object other = p.callFunction("get_other");
auto o = p.callFunction("py_test", other);
assertEq(1, o);
}
{
PythonProgram p("
from threading import Thread
import qoreloader
from qore.__root__.Qore.Thread import Queue
from qore.__root__.QTest import QTest
class Err(QTest):
def test(self) -> None:
self.raise_err()
def do_test() -> None:
err = Err()
err.test()
def test() -> None:
do_test()
return 1
", "test.py");
assertThrows("1", \p.callFunction(), "test");
}
}
superTest() {
{
PythonProgram p("
import qoreloader
from qore.__root__.Qore.Thread import Sequence
class py_sequence(Sequence):
def __init__(self):
pass
def test():
return py_sequence()
", "test.py");
auto o = p.callFunction("test");
assertNothing(o);
}
{
PythonProgram p("
import qoreloader
from qore.__root__.Qore.Thread import Sequence
class py_sequence(Sequence):
def __init__(self):
if (self.getCurrent() > 1):
raise Exception('err', 'desc')
def test():
return py_sequence()
", "test.py");
assertThrows("builtins.ValueError", \p.callFunction(), "test");
}
{
PythonProgram p("
import qoreloader
from qore.__root__.Qore.Thread import Sequence
class py_sequence(Sequence):
def __init__(self):
super(py_sequence, self).__init__()
if (self.getCurrent() > 1):
raise Exception('err', 'desc')
def test():
return py_sequence()
", "test.py");
assertEq(0, p.callFunction("test").getCurrent());
}
}
exceptionTest() {
PythonProgram p("
def t1():
t2()
def t2():
t3()
def t3():
raise Exception('err', 'desc')
", "test.py");
try {
p.callFunction("t1");
assertTrue(False);
} catch (hash ex) {
assertEq("t3", ex.callstack[0].function);
assertEq("t2", ex.callstack[1].function);
assertEq("t1", ex.callstack[2].function);
}
}
qoreTest() {
# Uses background threads which aren't compatible with free-threading mode
if (Python::FreeThreading) {
testSkip("Background threads with Python not supported with free-threading (PEP 703)");
}
{
PythonProgram::evalStatement("from qore.__root__.Qore import Serializable", "test.py");
assertEq(1, PythonProgram::evalExpression("Serializable.deserialize(Serializable.serialize(1))"));
}
{
hash object_cache;
code callback = sub (object obj) {
# save object in object cache, so it doesn't go out of scope
object_cache{obj.uniqueHash()} = obj;
};
PythonProgram p("", "");
p.importNamespace("::Test", "x");
p.setSaveObjectCallback(callback);
p.evalStatementKeep("
from qore.__root__.Test import Other
class PyTest:
def __init__(self):
self.mydict = {
'other': Other(),
}
def get(self, attr, meth, *argv):
return getattr(self.mydict[attr], meth)(*argv)
def test():
return PyTest()", "test.py");
p.setSaveObjectCallback();
object obj = p.callFunction("test");
assertEq(1, obj.get("other", "get"));
Queue q();
background q.push(obj.get("other", "get"));
assertEq(1, q.get());
}
{
Program p(PO_NEW_STYLE);
p.issueModuleCmd("python", "parse python.py #
class PythonBase:
def get(self) -> int:
return 1
");
p.issueModuleCmd("python", "export-class PythonBase");
p.parse("
class QoreTest inherits PythonBase {
int get(int i) {
return PythonBase::get() + i;
}
}
QoreTest sub get_object() {
return new QoreTest();
}
", "qore");
object obj = p.callFunction("get_object");
assertEq(2, obj.get(1));
}
}
javaTest() {
# JNI is not compatible with Python free-threading mode due to mimalloc conflicts
if (Python::FreeThreading) {
testSkip("JNI not supported with Python free-threading (PEP 703)");
}
# Try to load jni module dynamically
try {
load_module("jni");
} catch () {
testSkip("no jni module");
}
if (ENV.QORE_JNI) {
{
# Java import test
PythonProgram p("
import qoreloader
qoreloader.issue_module_cmd('jni', 'add-classpath " + ENV.QORE_JNI + "')
from java.org.qore.lang.restclient import RestClient
def test(url):
return RestClient({'url': url}, True)
", "test.py");
string url = "https://localhost:8011/";
object obj = p.callFunction("test", url);
assertEq(url, obj.getURL());
}
{
# Java import test
PythonProgram p("
import qoreloader
qoreloader.issue_module_cmd('jni', 'add-classpath " + ENV.QORE_JNI + "')
from java.org.qore.lang import restclient
def test(url):
return restclient.RestClient({'url': url}, True)
", "test.py");
string url = "https://localhost:8011/";
object obj = p.callFunction("test", url);
assertEq(url, obj.getURL());
}
}
{
PythonProgram p("
import qoreloader
from java.org.qore.jni import Hash
def test():
return Hash()
", "test.py");
object h = p.callFunction("test");
h.put("a", 1);
assertEq(1, h.get("a"));
}
{
PythonProgram p("
import qoreloader
qoreloader.load_java('org.qore.jni.Hash')
from qore.__root__.Jni.org.qore.jni import Hash
def test():
return Hash()
", "test.py");
object h = p.callFunction("test");
h.put("a", 1);
assertEq(1, h.get("a"));
}
}
stackTest() {
# Uses background threads which aren't compatible with free-threading mode
if (Python::FreeThreading) {
testSkip("Background threads with Python not supported with free-threading (PEP 703)");
}
PythonProgram p("
def f():
return f()", "");
assertThrows("builtins.RecursionError", \p.callFunction(), "f");
assertThrows("builtins.RecursionError", sub () {
Queue q();
code doit = sub () {
try {
p.callFunction("f");
q.push();
} catch (hash ex) {
q.push(ex);
}
};
background doit();
*hash ex = q.get();
if (ex) {
throw ex.err, ex.desc, ex.arg;
}
});
}
threadTest() {
# Multi-threaded access to Python sub-interpreters not supported with free-threading
# due to mimalloc heap initialization issues
if (Python::FreeThreading) {
testSkip("Multi-threaded Python access not supported with free-threading (PEP 703)");
}
{
string src = "
class test:
def __init__(self):
self.a = 0
self.cnt = 0
def test(self, v):
for i in range(10):
self.a += v
self.cnt += 1
def info(self):
return {
'a': self.a,
'cnt': self.cnt,
}
def get():
return test()
";
PythonProgram p0(src, "test.py");
PythonProgram p1(src, "test.py");
object py_obj0 = p0.callFunction("get");
object py_obj1 = p1.callFunction("get");
ThreadPool tp();
code test0 = sub () {
map py_obj0.test(1), xrange(10);
};
code test1 = sub () {
map py_obj1.test(1), xrange(10);
};
map tp.submit(test0), xrange(10);
map tp.submit(test1), xrange(10);
tp.stopWait();
hash info = py_obj0.info();
assertEq(1000, info.a);
assertEq(1000, info.cnt);
info = py_obj1.info();
assertEq(1000, info.a);
assertEq(1000, info.cnt);
}
{
string py_src = "
from qore.__root__ import TestApi, SubApi
class PySub(SubApi):
def __init__(self, parent):
super(PySub, self).__init__(parent)
def callbackImpl(self, msg):
return self.parent.process('py-' + msg)
class PyTest(TestApi):
def __init__(self):
super(PyTest, self).__init__()
def getImpl(self):
return PySub(self)
";
Program p0(PO_NEW_STYLE);
p0.importClass("TestApi");
p0.importClass("SubApi");
p0.issueModuleCmd("python", "parse python.py " + py_src);
p0.issueModuleCmd("python", "export-class PyTest");
Program p1(PO_NEW_STYLE);
p1.importClass("TestApi");
p1.importClass("SubApi");
p1.issueModuleCmd("python", "parse python.py " + py_src);
p1.issueModuleCmd("python", "export-class PyTest");
Class c0 = Class::forName(p0, "PyTest");
Class c1 = Class::forName(p1, "PyTest");
TestApi testapi0 = c0.newObject();
TestApi testapi1 = c1.newObject();
ThreadPool tp();
int cnt = 0;
code test0 = sub () {
SubApi s = testapi0.get();
map cnt += s.callback("msg"), xrange(100);
};
code test1 = sub () {
SubApi s = testapi1.get();
map cnt += s.callback("msg"), xrange(100);
};
map tp.submit(test0), xrange(10);
map tp.submit(test1), xrange(10);
tp.stopWait();
assertEq(12000, cnt);
}
{
PythonProgram p("
from qore.HttpServer import HttpServer
from qore.WebSocketHandler import WebSocketHandler, WebSocketConnection
from qore.__root__.Qore import printf, vprintf
class PyWsConnection(WebSocketConnection):
def __init__(self, handler):
super(PyWsConnection, self).__init__(handler)
def gotMessage(self, msg):
self.send('default response')
class MyWebSocketHandler(WebSocketHandler):
def __init__(self):
super(MyWebSocketHandler, self).__init__()
def getConnectionImpl(self, cx, hdr, cid):
return PyWsConnection(self)
def logError(self, fmt, *args):
vprintf(fmt + '\\n', args)
def log(fmt, *args):
#vprintf(fmt + '\\n', args)
pass
hs = HttpServer(log, log)
ws = MyWebSocketHandler()
hs.setDefaultHandler('ws', ws)
port = hs.addListener(0)['port']
def get_port():
global port
return port
def stop():
global hs
hs.stop()
", "ws.py");
int port = p.callFunction("get_port");
assertGt(0, port);
Queue q();
code callback = sub (data msg) {
q.push(msg);
};
WebSocketClient wsc(callback, {"url": "ws://localhost:" + port});
wsc.connect();
on_exit wsc.disconnect();
wsc.send("test");
assertEq("default response", q.get(10s));
# stop() must be called from the main thread (Gate owner), but it
# waits for all connections to drain first. Disconnect the client
# in a background thread so the main thread's stop() can complete.
# The main thread reaches p.callFunction("stop") before the new
# thread starts due to thread-creation overhead.
background wsc.disconnect();
p.callFunction("stop");
}
}
pythonCallbackTest() {
# Uses background threads which aren't compatible with free-threading mode
if (Python::FreeThreading) {
testSkip("Background threads with Python not supported with free-threading (PEP 703)");
}
PythonProgram p("
class c:
def __init__(self):
self.d = {
'x': 1,
}
def get(self):
return self.t
def t(self):
return self.d['x']
@classmethod
def tc():
return 2
@staticmethod
def sm():
return 3
def get_method():
c0 = c()
return c0.get()
def get_cm():
return c.tc
def get_sm():
return c.sm
def test_code(code):
code(1)
", "test.py");
code c = p.callFunction("get_method");
assertEq(1, c());
Queue q();
code bg = auto sub (code c) {
q.push(c());
};
background bg(c);
assertEq(1, q.get());
c = p.callFunction("get_cm");
assertEq(2, c());
background bg(c);
assertEq(2, q.get());
c = p.callFunction("get_sm");
assertEq(3, c());
background bg(c);
assertEq(3, q.get());
int i;
code inc = sub(int d) {
i += d;
};
p.callFunction("test_code", inc);
assertEq(1, i);
}
pythonImportTest() {
# Uses background threads which aren't compatible with free-threading mode
if (Python::FreeThreading) {
testSkip("Background threads with Python not supported with free-threading (PEP 703)");
}
{
PythonProgram p("", "");
p.importNamespace("::Test", "x");
p.evalStatementKeep("
class PyTest(x.TestClass):
def __init__(self, i):
super(PyTest, self).__init__(i)
def pyget(self):
return self.get()
def test(*args):
return PyTest(*args)", "test.py");
object obj = p.callFunction("test", 1);
assertEq(1, obj.pyget());
assertEq(1, obj.i);
# test with no args to base class
assertThrows("builtins.TypeError", \p.callFunction(), "test");
# test with too many args to Qore base class
assertThrows("builtins.TypeError", \p.callFunction(), ("test", 1, 2));
}
{
PythonProgram p("from qore.__root__ import Test as t
def get():
return t.get()", "test.py");
assertEq("55.00 €", p.callFunction("get"));
}
{
Program p(PO_NEW_STYLE);
p.issueModuleCmd("python", "import-ns Test test");
p.issueModuleCmd("python", "parse t from test import X");
assertEq(Type::Object, p.type(), "program is valid after import-ns and parse");
}
{
Program p(PO_NEW_STYLE);
p.issueModuleCmd("python", "import-ns Test test");
p.issueModuleCmd("python", "parse t from test.X import T");
assertEq(Type::Object, p.type(), "program is valid after nested parse");
}
{
Program p(PO_NEW_STYLE);
p.issueModuleCmd("python", "import-ns Test test");
p.issueModuleCmd("python", "alias test.X x");
p.issueModuleCmd("python", "parse t from x import T");
assertEq(Type::Object, p.type(), "program is valid after alias and parse");
}
{
PythonProgram p("
def test1():
return x.ex1()
def test2():
return x.ex2()
def z():
return y.z.exz()
", "test.py");
p.importNamespace("::Test", "x");
assertThrows("1", \p.callFunction(), ("test1"));
assertThrows("TEST", "test", \p.callFunction(), ("test2"));
p.aliasDefinition("x.ex1", "y.z.exz");
assertThrows("1", \p.callFunction(), ("z"));
p.evalStatementKeep("
class PyTest(x.Test):
def py_test(self):
return 2
", "new.py");
p.evalStatementKeep("
def py_test_func(i):
t = PyTest()
if (i == 1):
return t.get()
return t.py_test()
", "new2.py");
assertEq(2, p.callFunction("py_test_func", 0));
assertEq(1, p.callFunction("py_test_func", 1));
code call_bg = auto sub (string func) {
Queue q();
*list args = argv;
background (sub () {
try {
q.push(p.callFunctionArgs(func, args));
} catch (hash ex) {
q.push(ex);
}
})();
return q.get();
};
assertEq(2, call_bg("py_test_func", 0));
assertEq(1, call_bg("py_test_func", 1));
p.evalStatementKeep("
class PyAbstractTest(x.AbstractTest):
def __init__(self, other):
super(PyAbstractTest, self).__init__(other)
def get(self):
return 3
def getpy(self):
return 4
def get_other(self):
return self.other.get()
class PyOther(x.Other):
def __init__(self):
super(PyOther, self).__init__()
def get2(self):
return self.get() + 1
def get3(self):
return self.other2.get()
", "new3.py");
p.evalStatementKeep("
def py_abstract_test(x = 0):
t = PyAbstractTest(None)
if (x):
return t.getpy()
return t.get()
def py_get_abstract(other):
return PyAbstractTest(other)
def py_get_other():
return PyOther()
", "new4.py");
assertEq(3, p.callFunction("py_abstract_test"));
assertEq(4, p.callFunction("py_abstract_test", 1));
assertEq(3, call_bg("py_abstract_test"));
assertEq(4, call_bg("py_abstract_test", 1));
Other other = p.callFunction("py_get_other");
assertEq(2, cast