#include <string>
#include <cstdio>
#include <opencog/atomspace/AtomSpace.h>
#include <opencog/cython/PythonEval.h>
#include <opencog/guile/SchemeEval.h>
#include <cxxtest/TestSuite.h>
using std::string;
using namespace opencog;
const int NO_SIGNAL_HANDLERS = 0;
class PythonEvalUTest : public CxxTest::TestSuite
{
public:
PythonEvalUTest()
{
logger().set_print_to_stdout_flag(true);
// setup global logger
logger().set_level(Logger::DEBUG);
logger().set_print_to_stdout_flag(true);
}
~PythonEvalUTest() {}
void setUp() {}
void tearDown() {}
void testRawPythonInitialization()
{
logger().debug("[PythonEvalUTest] testRawPythonInitialization()");
// Start up Python.
Py_InitializeEx(NO_SIGNAL_HANDLERS);
#if PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION < 9
PyEval_InitThreads();
#endif
// Release the GIL.
// PyEval_ReleaseLock();
// Stop Python.
Py_Finalize();
// Try it again to make sure the Python library cleans up properly.
// Start up Python.
Py_InitializeEx(NO_SIGNAL_HANDLERS);
#if PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION < 9
PyEval_InitThreads();
#endif
// Release the GIL.
// PyEval_ReleaseLock();
// Stop Python.
Py_Finalize();
// And again...
// Start up Python.
Py_InitializeEx(NO_SIGNAL_HANDLERS);
#if PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION < 9
PyEval_InitThreads();
#endif
// Release the GIL.
// PyEval_ReleaseLock();
// Stop Python.
Py_Finalize();
logger().debug("[PythonEvalUTest] testRawPythonInitialization() DONE");
}
void testRawPythonInitializeWithGetSys()
{
logger().debug("[PythonEvalUTest] testRawPythonInitializeWithGetSys()");
PyObject* pySysPath = NULL;
// Get the system path object.
Py_InitializeEx(NO_SIGNAL_HANDLERS);
#if PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION < 9
PyEval_InitThreads();
#endif
PyRun_SimpleString(
"import sys\n"
"import io\n"
);
pySysPath = PySys_GetObject((char*)"path");
if (logger().is_debug_enabled()) {
logger().debug(" unit-test Python 'sys.path' is:");
Py_ssize_t pathSize = PyList_Size(pySysPath);
for (int pathIndex = 0; pathIndex < pathSize; pathIndex++) {
PyObject* pySysPathLine = PyList_GetItem(pySysPath, pathIndex);
PyObject* pyStr = nullptr;
if (not PyBytes_Check(pySysPathLine)) {
pyStr = PyUnicode_AsEncodedString(pySysPathLine,
"UTF-8", "strict");
pySysPathLine = pyStr;
}
const char* sysPathCString = PyBytes_AsString(pySysPathLine);
logger().debug(" %2d > %s", pathIndex, sysPathCString);
// NOTE: PyList_GetItem returns borrowed reference so don't do this:
// Py_DECREF(pySysPathLine);
if (pyStr) Py_DECREF(pyStr);
}
}
// NOTE: PySys_GetObject returns a borrowed reference so don't do this:
// Py_DECREF(pySysPath);
// PyEval_ReleaseLock();
Py_Finalize();
// Get it again
Py_InitializeEx(NO_SIGNAL_HANDLERS);
#if PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION < 9
PyEval_InitThreads();
#endif
PyRun_SimpleString(
"import sys\n"
"import io\n"
);
pySysPath = PySys_GetObject((char*)"path");
// NOTE: PySys_GetObject returns a borrowed reference so don't do this:
// Py_DECREF(pySysPath);
// PyEval_ReleaseLock();
Py_Finalize();
logger().debug("[PythonEvalUTest] testRawPythonInitializeWithGetSys() DONE");
}
void testPythonEvalSingletonCreationDeletion()
{
logger().debug("[PythonEvalUTest] testPythonEvalSingletonCreationDeletion()");
// Initialize Python.
global_python_initialize();
// Test evaluator pool creation
AtomSpacePtr as1 = createAtomSpace();
PythonEval* evaluator1 = PythonEval::get_python_evaluator(as1);
TS_ASSERT(evaluator1 != nullptr);
// Get evaluator for different atomspace
AtomSpacePtr as2 = createAtomSpace();
PythonEval* evaluator2 = PythonEval::get_python_evaluator(as2);
TS_ASSERT(evaluator2 != nullptr);
// Getting evaluator for same atomspace should return same instance
PythonEval* evaluator1_again = PythonEval::get_python_evaluator(as1);
TS_ASSERT(evaluator1 == evaluator1_again);
logger().debug("[PythonEvalUTest] testPythonEvalSingletonCreationDeletion() DONE");
}
void testPythonEvalEvalExpr()
{
logger().debug("[PythonEvalUTest] testPythonEvalEvalExpr()");
AtomSpacePtr as = createAtomSpace();
PythonEval* evaluator = PythonEval::get_python_evaluator(as);
// Check that our atom space global is visible and can add a node.
evaluator->clear_pending();
evaluator->eval_expr("from opencog.atomspace import *");
evaluator->eval_expr("ATOMSPACE = AtomSpace()");
evaluator->eval_expr("ATOMSPACE.add_node(types.ConceptNode,'test')");
evaluator->eval_expr("print ('Total atoms = ', ATOMSPACE.size(), '\\n')");
logger().debug("[PythonEvalUTest] testPythonEvalEvalExpr() DONE");
}
void testApplyAndApplyTV()
{
AtomSpacePtr as = createAtomSpace();
PythonEval* python = PythonEval::get_python_evaluator(as);
// No. Throwing messes up the cogserver. Just report this as
// an ordinary syntax error.
//TS_ASSERT_THROWS(std::string eval_res = python->eval(
// "from opencog.atomspace import vsfvkdfjnv\n"), RuntimeException);
// Define python functions for use by scheme.
std::string eval_res = python->eval(
"from opencog.atomspace import Atom\n"
"def add_link(atom1, atom2):\n"
" ATOMSPACE = atom1.atomspace\n"
" link = ATOMSPACE.add_link(types.ListLink, [atom1, atom2])\n"
" print ('add_link(',atom1,',',atom2,') = ', link)\n"
" return link\n\n"
);
std::cout << "eval_res:" << std::endl << eval_res << std::endl;
#ifdef HAVE_GUILE
// Create a scheme evaluator to trigger cog-execute.
SchemeEval* scheme = SchemeEval::get_scheme_evaluator(as);
// Test execute -> PythonEval::apply
scheme->eval_h(
"(cog-execute! "
" (ExecutionOutputLink"
" (GroundedSchemaNode \"py: add_link\")"
" (ListLink"
" (ConceptNode \"one\")"
" (ConceptNode \"two\")"
" )"
" )"
")"
);
bool error = scheme->eval_error();
scheme->clear_pending();
TSM_ASSERT("Failed cog-execute!", !error);
#endif // HAVE_GUILE
}
void testCodeBlockWithNewline()
{
AtomSpacePtr as = createAtomSpace();
PythonEval* python = PythonEval::get_python_evaluator(as);
// Define python functions with newline within the code block
string result = python->eval(
"def fun(x):\n"
" y = x + 1\n"
"\n"
" return y\n"
"\n"
"fun(1)\n\n"
);
std::cout << "result = " << result << std::endl;
// Because results are ignored this cannot be tested as of now
// TS_ASSERT_EQUALS(result, "3");
TS_ASSERT(true);
}
void testCodeBlockWithNewlineThreaded()
{
AtomSpacePtr as = createAtomSpace();
PythonEval* python = PythonEval::get_python_evaluator(as);
std::vector<std::thread> threads;
std::string s =
"tmp = 0\n"
"def fun(x):\n"
" y = x + 1\n"
"\n"
" global tmp\n"
" tmp += x\n"
" return y\n"
"\n"
"print(tmp)\n"
"fun(1)\n\n";
std::string s2 = "fun(1)\n\n";
python->eval(s);
for(int i=0; i<10; i++){
// Define python functions with newline within the code block
threads.emplace_back(&PythonEval::eval, python,
std::ref(s2)
);
}
for(std::thread & t: threads)
t.join();
std::string result = python->eval("print(fun(15))\n");
std::cout << "result = " << result << std::endl;
result = python->eval("print(tmp)\n");
std::cout << "result = " << result << std::endl;
// Because results are ignored this cannot be tested as of now
//TS_ASSERT_EQUALS(result, "26");
TS_ASSERT(true);
}
void testGlobalPythonInitializationFinalization()
{
logger().debug("[PythonEvalUTest] testGlobalPythonInitializationFinalization()");
// Initialize and finalize Python.
global_python_initialize();
global_python_finalize();
global_python_initialize();
global_python_finalize();
global_python_initialize();
global_python_finalize();
logger().debug("[PythonEvalUTest] testGlobalPythonInitializationFinalization() DONE");
}
};