/*
* tests/shell/ShellUTest.cxxtest
*
* Copyright (C) 2016 Linas Vepstas
* All Rights Reserved
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License v3 as
* published by the Free Software Foundation and including the exceptions
* at http://opencog.org/wiki/Licenses
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program; if not, see http://www.gnu.org/licenses/
*/
#include <thread>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <time.h>
#include <opencog/atomspace/AtomSpace.h>
#include <opencog/atoms/atom_types/atom_names.h>
#include <opencog/atoms/value/FloatValue.h>
#include <opencog/atoms/value/VoidValue.h>
#include <opencog/cogserver/atoms/CogServerNode.h>
using namespace opencog;
std::string cmd_exec(const char* cmd)
{
#define BUFSZ 1000
char buf[1000];
std::string result;
std::shared_ptr<FILE> pope(popen(cmd, "r"), pclose);
if (!pope) throw std::runtime_error("popen() failed!");
while (!feof(pope.get())) {
if (fgets(buf, BUFSZ, pope.get()) != NULL)
result += buf;
}
return result;
}
void gargleblast(int tid, int reps)
{
for (int i=0; i<reps; i++)
{
int j = i + 10 + (tid+10)*1000;
char buf[1000];
sprintf(buf, "printf '(Evaluation (Predicate \"visible face\") (List (Number %d)))\\n' | nc -q 1 localhost 17333", j);
std::string reso = cmd_exec(buf);
// The current reply message is exactly:
// opencog> Entering scheme shell; use ^D or a single . on a line by itself to exit.
// guile> (EvaluationLink
// (PredicateNode "visible face")
// (ListLink
// (NumberNode "28010")))
//
// guile>
// guile> ; note blank space here
// Which is exactly 266 chars long.
// If the greeting changes, then adjust for the correct size.
//
// Hmm. Seems that sometimes some trailing newline gets cut,
// and so the reply might be short by one byte. Seems harmless.
//
// The new half-open-detection code will transmit spurious
// blank spaces. We are going to ignore those spaces entirely.
int nch = 0;
for (size_t i=0; i<reso.size(); i++)
// if (' ' != reso[i]) nch++;
if (0x16 != reso[i]) nch++;
// #define CUR_MSG_SZ 235 // Size without blank spaces.
// #define CUR_MSG_SZ 264 // Size without synchronous idle.
// #define CUR_MSG_SZ 248 // Size without 'Node", 'Link' suffixes.
#define CUR_MSG_SZ 223 // Size without extra colorized `guile>`
// size_t msgsz = reso.size();
size_t msgsz = nch;
TS_ASSERT_LESS_THAN_EQUALS(CUR_MSG_SZ, msgsz);
TS_ASSERT_LESS_THAN_EQUALS(msgsz, CUR_MSG_SZ+1);
if (msgsz < CUR_MSG_SZ or (CUR_MSG_SZ+1) < msgsz)
printf("Error: Create rep %d of %d in thread %d failed\n"
"\tGot: %zu >>>%s<<<\n",
i, reps, tid, reso.size(), reso.c_str());
usleep(50000);
if (20 <= i)
{
sprintf(buf, "printf '(cog-extract! (Evaluation (Predicate \"visible face\") (List (Number %d))))\\n' | nc -q 1 localhost 17333", j-20);
std::string delo = cmd_exec(buf);
// The new half-open detection code sends blanks.
// Ignore them by noot counting them.
int nch = 0;
for (size_t i=0; i<delo.size(); i++)
// if (' ' != delo[i]) nch++;
if (0x16 != delo[i]) nch++;
// #define CUR_DEL_SZ 160 // Size without blank spaces
// #define CUR_DEL_SZ 178 // Size without synchronous idle
#define CUR_DEL_SZ 153 // Size without extra colorized `guile>`
// size_t msgsz = delo.size();
size_t msgsz = nch;
TS_ASSERT_LESS_THAN_EQUALS(CUR_DEL_SZ, msgsz);
TS_ASSERT_LESS_THAN_EQUALS(msgsz, CUR_DEL_SZ+1);
if (msgsz < CUR_DEL_SZ or (CUR_DEL_SZ+1) < msgsz)
printf("Error: Delete rep %d of %d in thread %d failed\n"
"\tGot: %zu >>>%s<<<\n",
i, reps, tid, delo.size(), delo.c_str());
}
usleep(50000);
}
}
// Test sending and receiving multiple commands on the same socket.
void tik_tik_tik(int tid, int reps)
{
static bool already_printed = false;
// Open a socket to the server.
int sock = socket(AF_INET, SOCK_STREAM, 0);
TS_ASSERT(0 < sock);
struct sockaddr_in server;
server.sin_addr.s_addr = inet_addr("127.0.0.1");
server.sin_family = AF_INET;
server.sin_port = htons(17333);
int rc = connect(sock, (struct sockaddr *)&server, sizeof(server));
TS_ASSERT(0 == rc);
// Put the shell into scheme-mode.
rc = send(sock, "scm\n", strlen("scm\n"), 0);
TS_ASSERT(4 == rc);
char srep[200];
rc = recv(sock, srep, sizeof(srep), 0);
TS_ASSERT(0 < rc);
// A send+receive loop, reusing the above socket.
for (int i=0; i<reps; i++)
{
char buf[200];
sprintf(buf, "(function-1 \"TICKET-%d-%d\")\n", tid+5000, i+8000);
rc = send(sock, buf, strlen(buf), 0);
// Receive loop. Stalls until all of the answer has been
// received. We know that we got all of it, if the string
// "finished" appears in the answer.
int count = 0;
std::string reply;
while (1) {
char rep[1000];
rc = recv(sock, rep, sizeof(rep), 0);
int norr = errno;
if (rc < 0 and EAGAIN != norr and EWOULDBLOCK != norr)
{
printf("Ohhh Nooo! %d %s\n", norr, strerror(norr));
TS_ASSERT(0);
break;
}
if (0 < rc) { rep[rc] = 0; reply += rep; }
// Oops. Fatty boom boom.
TS_ASSERT(reply.npos == reply.find("ERROR"));
if (reply.npos != reply.find("ERROR"))
{
printf("Failure on thread-id=%d rep=%d\n", tid, i);
if (not already_printed)
printf("Shell reply is: %s\n", reply.c_str());
already_printed = true;
close(sock);
return;
}
// Look for zef side.
if (reply.npos != reply.find("-finished")) break;
count ++;
if (100 < count) break;
}
char exp[200];
sprintf(exp, "return-1-TICKET-%d-%d-finished", tid+5000, i+8000);
if (reply.npos == reply.find(exp))
printf("Failed to get expected reply; \n"
"expected: >>%s<<\n"
"got >>%s<<\n", exp, reply.c_str());
TS_ASSERT(reply.npos != reply.find(exp));
}
// Be polite, and close up shop.
close(sock);
}
class ShellUTest : public CxxTest::TestSuite
{
private:
AtomSpacePtr asp;
CogServerNodePtr csrv;
public:
ShellUTest()
{
// logger().set_level(Logger::DEBUG);
logger().set_level(Logger::INFO);
// logger().set_print_to_stdout_flag(true);
asp = createAtomSpace();
Handle hcsn = asp->add_node(COG_SERVER_NODE, "test-cogserver");
csrv = CogServerNodeCast(hcsn);
// Set ports and start the server
csrv->setValue(asp->add_atom(Predicate("*-telnet-port-*")),
createFloatValue(17333.0));
csrv->setValue(asp->add_atom(Predicate("*-web-port-*")),
createFloatValue(0.0));
csrv->setValue(asp->add_atom(Predicate("*-mcp-port-*")),
createFloatValue(0.0));
csrv->setValue(asp->add_atom(Predicate("*-start-*")),
createVoidValue());
}
~ShellUTest()
{
logger().debug("AFTER TEST stop cogserver: %s", __FUNCTION__);
csrv->setValue(asp->add_atom(Predicate("*-stop-*")),
createVoidValue());
csrv = nullptr;
asp = nullptr;
logger().debug("AFTER TEST DONE stopping cogserver: %s", __FUNCTION__);
// erase the log file if no assertions failed
if (!CxxTest::TestTracker::tracker().suiteFailed())
std::remove(logger().get_filename().c_str());
}
void setUp()
{
}
void tearDown()
{
}
void testControlD()
{
time_t t0 = time(0);
logger().debug("BEGIN TEST: %s", __FUNCTION__);
int rc = system("printf '(Concept \"42\")\\n' | nc -q 0 localhost 17333");
if (rc != 0) printf("Error: first call failed\n");
TS_ASSERT_EQUALS(rc, 0);
usleep(50000);
// The hexadecimal \x4 is ctrl-D.
rc = system("printf 'scm\\n(+ 2 2)\\n\x4\\n' | nc -q 0 localhost 17333");
if (rc != 0) printf("Error: second call failed\n");
TS_ASSERT_EQUALS(rc, 0);
usleep(50000);
rc = system("printf 'scm\\n(Concept \"42\")\\n' | nc -q 0 localhost 17333");
if (rc != 0) printf("Error: third call failed\n");
TS_ASSERT_EQUALS(rc, 0);
usleep(50000);
rc = system("printf '(+ 2 2)\\n.\\n' | nc -q 0 localhost 17333");
if (rc != 0) printf("Error: fourth call failed\n");
TS_ASSERT_EQUALS(rc, 0);
usleep(50000);
printf("TEST testControlD: %ld seconds\n", time(0) - t0);
logger().debug("END TEST: %s", __FUNCTION__);
}
void testControlC()
{
time_t t0 = time(0);
logger().debug("BEGIN TEST: %s", __FUNCTION__);
time_t start = time(0);
// The hexadecimal \x16 is ctrl-C.
// The interrupt should kill the sleep immediately.
// Elapsed time should be well under a second.
int rc = system("printf '(sleep 42)\\n\x16' | nc -q 0 localhost 17333");
if (rc != 0) logger().warn("Error: first call failed");
TS_ASSERT_EQUALS(rc, 0);
time_t end = time(0);
TS_ASSERT_LESS_THAN (end - start, 2);
logger().debug("Elapsed seconds: %ld", end-start);
start = time(0);
// The hexadecimal \x18 is ctrl-X aka "cancel"
rc = system("printf '(sleep 43)\\n\x18' | nc -q 0 localhost 17333");
if (rc != 0) logger().warn("Error: first call failed");
TS_ASSERT_EQUALS(rc, 0);
end = time(0);
TS_ASSERT_LESS_THAN (end - start, 2);
logger().debug("First elapsed: %ld", end-start);
start = time(0);
// The hexadecimal \x1b is ctrl-[ aka "escape"
rc = system("printf '(sleep 44)\\n\x1b' | nc -q 0 localhost 17333");
if (rc != 0) logger().warn("Error: second call failed");
TS_ASSERT_EQUALS(rc, 0);
end = time(0);
TS_ASSERT_LESS_THAN (end - start, 2);
logger().debug("Second elapsed: %ld", end-start);
usleep(50000);
printf("TEST testControlC: %ld seconds\n", time(0) - t0);
logger().debug("END TEST: %s", __FUNCTION__);
}
// Test draining of the shell command queue, before the shell is
// closed. Here, net-cat sends some long-running commands, but
// then closes the socket immediately. All of the commands should
// still run to completion, however.
void testDrain()
{
time_t t0 = time(0);
logger().debug("BEGIN TEST: %s", __FUNCTION__);
cmd_exec(
"printf 'scm\\n (sleep 3) (Concept \"A\")\\n"
"(sleep 3)\\n (Concept \"B\")\\n"
"(sleep 3)\\n (Concept \"C\")\\n' | nc -q 0 localhost 17333");
// The barrier call here will force above to finish before the
// atomspace print below. We sleep one second, to make sure the
// above got started; we don't want the barrier to finish before
// the above starts.
sleep(1);
cmd_exec("printf 'scm\\n (cog-global-barrier)' | nc -q 0 localhost 17333");
// Sleep one second, so as to let the barrier at least get started,
// before we ask to print the AtomSpace. The barrier should block
// for about nine seconds (the sum of the sleeps above). We won't
// notice the block, but the attempt to cog-prt-atomspace will.
// Use `nc -q 15` because it will take a while to respond.
sleep(1);
std::string ply = cmd_exec(
"printf 'scm\\n (cog-prt-atomspace)' | nc -q 15 localhost 17333");
// Expect between 189 and 205 bytes; it seems a trailing
// prompt string is sometimes not received. I don't know
// if that's OK, or if that's a bug. Below, we also test
// to make sure the printed AtomSpace is correct.
printf("Received %zu bytes: >>>%s<<<\n", ply.size(), ply.c_str());
TS_ASSERT_LESS_THAN(189, ply.size());
// Expect to get all three atoms.
TS_ASSERT(ply.npos != ply.find("(Concept \"A\")"));
TS_ASSERT(ply.npos != ply.find("(Concept \"B\")"));
TS_ASSERT(ply.npos != ply.find("(Concept \"C\")"));
printf("TEST testDrain: %ld seconds\n", time(0) - t0);
logger().debug("END TEST: %s", __FUNCTION__);
}
void testMessaging()
{
time_t t0 = time(0);
logger().debug("BEGIN TEST: %s", __FUNCTION__);
#define NT 20
std::thread* pangalactic[NT];
int j;
for (j=0; j<NT; j++)
{
pangalactic[j] = new std::thread(&gargleblast, j, 30);
}
for (j=0; j<NT; j++)
{
pangalactic[j]->join();
delete pangalactic[j];
}
printf("TEST testMessaging: %ld seconds\n", time(0) - t0);
logger().debug("END TEST: %s", __FUNCTION__);
}
// Test streaming commands onto shared sockets.
// Unlike the above test, here the socket is kept open
// indefinitely.
void testMultiStream()
{
time_t t0 = time(0);
logger().debug("BEGIN TEST: %s", __FUNCTION__);
std::string ply = cmd_exec("printf 'scm\\n"
// "(add-to-load-path \"/usr/local/share/opencog/scm\")\\n"
// "(use-modules (opencog) (opencog logger))"
"(define (function-1 ticket) "
// " (cog-logger-info \"CALLED function-1 ~s\" ticket)"
" (string-append \"return-1-\" ticket \"-finished\"))\\n"
"(define (function-2 ticket) "
// " (cog-logger-info \"CALLED function-2 ~s\" ticket)"
" (string-append \"return-2-\" ticket \"-finished\"))\\n"
"(display \"Finished creating definitions\\n\")"
"\\n.\\n' | nc -q 2 localhost 17333");
// The -q2 on `nc` should be enough for the above to finish up.
TS_ASSERT (0 < ply.size());
#define NTMS 130
#define NREPS 3000
std::thread* max_normal[NTMS];
int j;
for (j=0; j<NTMS; j++)
{
max_normal[j] = new std::thread(&tik_tik_tik, j, NREPS);
}
for (j=0; j<NTMS; j++)
{
max_normal[j]->join();
delete max_normal[j];
}
printf("TEST testMultiStream: %ld seconds\n", time(0) - t0);
logger().debug("END TEST: %s", __FUNCTION__);
}
};