PyFCGI/tests/check_monitor.c

#include <stdlib.h>
#include <check.h>
#include <stdio.h>

#include "monitor.h"

static const char *valids[] = {
	"tcp://localhost:8181",
	"tcp://127.0.0.1:81",
	"udp://localhost:8000",
	"udp://192.168.13.37:1234",
	//"tcp://::1:8000",
	"unix:///tmp/test/test.sock",
	"unix://foo.sock",
	NULL};

static const char *valids6[] = {
	"tcp://localhost:8181",
	"udp://localhost:8000",
	//"tcp://::1:8000",
	"unix:///tmp/test/test.sock",
	"unix://foo.sock",
	NULL};

static const char *invalids[] = {
	"localhost:8000",
	"ttt://127.0.0.1:81",
	"tcp://","udp://", "unix://",
	"tcp://localhost",
	};

START_TEST(test_check_sock_valid)
{
	int ret;
	const char **urls;
	char err_msg[256];

	for(urls=valids; *urls != NULL; urls++)
	{
		ret = pyfcgi_monitor_check_sock(*urls);
		snprintf(err_msg, 256,
			"returns %d != 0 but '%s' given",
			ret, *urls);
		ck_assert_msg(ret == 0, err_msg);
	}
}
END_TEST

START_TEST(test_check_sock_invalid)
{
	int ret;
	const char **urls;
	char err_msg[256];

	for(urls=invalids; *urls != NULL; urls++)
	{
		ret = pyfcgi_monitor_check_sock(*urls);
		snprintf(err_msg, 256,
			"returns %d != -1 but '%s' given",
			ret, *urls);
		ck_assert_msg(ret == -1, err_msg);
	}
}
END_TEST

START_TEST(test_parse_sock_valid_ipv4)
{
	pyfcgi_monitor_addr_t addr;
	int domain, type, protocol, args[3], ret, i;
	const char **urls;
	char err_msg[256];

	const int expt_ret[] = {
		AF_INET, SOCK_STREAM, 0,
		AF_INET, SOCK_STREAM, 0,
		AF_INET, SOCK_DGRAM, 0,
		AF_INET, SOCK_DGRAM, 0,
		AF_UNIX, SOCK_STREAM, 0,
		AF_UNIX, SOCK_STREAM, 0,};
	const int *expts = expt_ret;

	PyFCGI_conf.ipv4 = 1;
	i=0;

	for(urls=valids; *urls != NULL; urls++)
	{
		ret = pyfcgi_monitor_parse_sock(*urls, args, &addr);
		snprintf(err_msg, 256,
			"returns %d != 0 but '%s' given",
			ret, *urls);
		ck_assert_msg(ret == 0, err_msg);

		domain = args[0];
		type = args[1];
		protocol = args[2];

		ck_assert_int_eq(expts[0], domain);
		ck_assert_int_eq(expts[1], type);
		ck_assert_int_eq(expts[2], protocol);

		/**@todo check returned address */
		expts+=3;
		i+=1;
	}
}
END_TEST

START_TEST(test_parse_sock_valid_ipv6)
{
	pyfcgi_monitor_addr_t addr;
	int domain, type, protocol, args[3], ret, i;
	const char **urls;
	char err_msg[256];

	const int expt_ret[] = {
		AF_INET6, SOCK_STREAM, 0,
		AF_INET6, SOCK_DGRAM, 0,
		AF_UNIX, SOCK_STREAM, 0,
		AF_UNIX, SOCK_STREAM, 0,};
	const int *expts = expt_ret;

	PyFCGI_conf.ipv6 = 1;
	i=0;

	for(urls=valids6; *urls != NULL; urls++)
	{
		ret = pyfcgi_monitor_parse_sock(*urls, args, &addr);
		snprintf(err_msg, 256,
			"returns %d != 0 but '%s' given",
			ret, *urls);
		ck_assert_msg(ret == 0, err_msg);

		domain = args[0];
		type = args[1];
		protocol = args[2];

		ck_assert_int_eq(expts[0], domain);
		ck_assert_int_eq(expts[1], type);
		ck_assert_int_eq(expts[2], protocol);

		/**@todo check returned address */
		expts+=3;
		i+=1;
	}
}
END_TEST


Suite * logger_suite(void)
{
	Suite *s;
	TCase *tc_sockurl;

	s = suite_create("Monitoring server");
	tc_sockurl = tcase_create("Socket URL handling functions");
	tcase_add_test(tc_sockurl, test_check_sock_valid);
	tcase_add_test(tc_sockurl, test_check_sock_invalid);
	tcase_add_test(tc_sockurl, test_parse_sock_valid_ipv4);
	tcase_add_test(tc_sockurl, test_parse_sock_valid_ipv6);

	suite_add_tcase(s, tc_sockurl);
	return s;
}

int main(void)
{
	int nfailed;
	Suite *s;
	SRunner *sr;

	s = logger_suite();
	sr = srunner_create(s);
	
	srunner_run_all(sr, CK_VERBOSE);
	nfailed = srunner_ntests_failed(sr);
	srunner_free(sr);

	return (!nfailed)?EXIT_SUCCESS:EXIT_FAILURE;
}