functional_test_qos.c

#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdarg.h>
#include <stddef.h>
#include <setjmp.h>
#include <errno.h>
#include <cmocka.h>

#include <libubus.h>
#include <libubox/blobmsg_json.h>
#include <libubox/blobmsg.h>

#include <json-validator.h>
#include <json-c/json.h>
#include <json-editor.h>
#include <json-c/json_tokener.h>

#include "qosmngr.h"

#define UBUS_OUTPUT_FILE "/tmp/qosmngr_ubus_output.txt"

struct json_object *json_output = NULL;

/**
 *  group_setup function to setup the test modules
 *  @param state input parameter pointer to a pointer to test state
 */
static int group_setup(void **state)
{
	// Start ubusd if it is not running
	system("pidof ubusd || ubusd &");

	// Restart qosmngr
	system("kill $(pidof qosmngr) 2>/dev/null; ../../qosmngr &");

	return 0;
}

/**
 *  group_teardown function to cleanup the test modules
 *  @param state input parameter pointer to a pointer to test state
 */
static int group_teardown(void **state)
{
	// Stop qosmngr
	system("kill $(pidof qosmngr) 2>/dev/null");

	// Delete test logs
	unlink(UBUS_OUTPUT_FILE);

	return 0;
}

/**
 *  teardown function to cleanup the test modules
 *  @param state input parameter pointer to a pointer to test state
 */
static int teardown(void **state)
{
	if (json_output) {
		json_object_put(json_output); // Free the output JSON object
		json_output = NULL;
	}

	return 0;
}

/**
 *  validate_queues function to validate all queues received under queues
 *  @param queues input parameter pointer to json_object containg all queues to
 *  be validated
 */
static void validate_queues(struct json_object *queues)
{
	int i;
	int len;

	assert_non_null(queues);
	len = json_object_array_length(queues);

	// Assert on length as well, should be greater than 0
	assert_true(len >= 0);

	for (i = 0; i < len; i++) {
		struct json_object *queue = json_object_array_get_idx(queues, i);
		json_object_object_foreach(queue, key, val) {
			int val_type = json_object_get_type(val);
			switch(val_type) {
			case json_type_int:
				assert_true(json_object_get_int(val) >= 0);
				break;
			case json_type_string:
				assert_true(!strncmp(json_object_get_string(val),
						   "eth", strlen("eth")));
				break;
			}
		}
	}
}

/**
 *  validate_stats function to validate json_object received under json_obj
 *  @param json_obj input parameter pointer to json_object containg ubus output
 */
static void validate_stats(struct json_object *json_obj)
{
        assert_non_null(json_obj);

	json_object_object_foreach(json_obj, key, val) {
		int val_type = json_object_get_type(val);

		if (val_type == json_type_array) {
			switch (val_type) {
			case json_type_array:
				validate_queues(val);
				break;
			}
		}
	}
}

/**
 *  get_ubus_call_output function to fill global json output variable with ubus call output
 *  @param ubus_cmd input parameter pointer to char string containg ubus call
 *  @param negative_case input parameter of type bool to identify negative scenarios
 */
static void get_ubus_call_output(const char *ubus_cmd, bool negative_case)
{
	char cmd[256];
	int fd = -1;
	char *str = NULL;

	// Put the output of ubus call into a string
	snprintf(cmd, sizeof(cmd), "%s > %s", ubus_cmd, UBUS_OUTPUT_FILE);
	int rc = system(cmd);
	assert_return_code(rc, errno);

	fd = open(UBUS_OUTPUT_FILE, O_RDONLY);
	assert_return_code(fd, errno);

	struct stat st;
	rc = fstat(fd, &st);

	if(negative_case) {
		assert_false(rc == 0 && st.st_size > 0);
		return;
	} else
		assert_true(rc == 0 && st.st_size > 0);

	str = calloc(1, (size_t)st.st_size + 1);
	assert_non_null(str);

	ssize_t read_len = read(fd, str, (size_t)st.st_size);
	assert_int_equal((int)read_len, (int)st.st_size);

	// Parse the string to a json_object
	//printf("JSON_OBJECT GET : %s\n", str);
	json_output = json_tokener_parse(str);

	if (fd >= 0)
		close(fd);
	if (str)
		free(str);
}

/**
 *  test_qos_stats_no_param function to test qos_stats without any parameter
 *  @param state input parameter pointer to a pointer to test state
 */
static void test_qos_stats_no_param(void **state)
{
	get_ubus_call_output("ubus call qos queue_stats", false);
	validate_stats(json_output);
}

/**
 *  test_qos_stats_one_param function to test qos_stats with just ifname
 *  @param state input parameter pointer to a pointer to test state
 */
static void test_qos_stats_one_param(void **state)
{
	get_ubus_call_output("ubus call qos queue_stats '{\"ifname\":\"eth0\"}'", false);
	validate_stats(json_output);
}

/**
 *  test_qos_stats_all_param function to test qos_stats with all valid params
 *  @param state input parameter pointer to a pointer to test state
 */
static void test_qos_stats_all_param(void **state)
{
	get_ubus_call_output("ubus call qos queue_stats '{\"ifname\":\"eth0\", \"qid\":0}'", false);
	validate_stats(json_output);
}

/**
 *  test_qos_stats_missing_param_ifname function to test missing parameter i.e., ifname
 *  @param state input parameter pointer to a pointer to test state
 */
static void test_qos_stats_missing_param_ifname(void **state)
{
	get_ubus_call_output("ubus call qos queue_stats '{\"qid\":0}'", true);
}

/**
 *  test_qos_stats_invalid_param_port function to test invalid parameter as port
 *  @param state input parameter pointer to a pointer to test state
 */
static void test_qos_stats_invalid_param_port(void **state)
{
	get_ubus_call_output("ubus call qos queue_stats '{\"port\":\"eth0\"}'", true);
}

/**
 *  test_qos_stats_invalid_param_pid function to test invalid parameter as pid
 *  @param state input parameter pointer to a pointer to test state
 */
static void test_qos_stats_invalid_param_pid(void **state)
{
	get_ubus_call_output("ubus call qos queue_stats '{\"ifname\":\"eth0\", \"pid\":0}'", true);
}

int main(void)
{
	const struct CMUnitTest tests[] = {
		cmocka_unit_test_teardown(test_qos_stats_no_param, teardown),
		cmocka_unit_test_teardown(test_qos_stats_one_param, teardown),
		cmocka_unit_test_teardown(test_qos_stats_all_param, teardown),

		// -ve scenarios
		cmocka_unit_test_teardown(test_qos_stats_missing_param_ifname, teardown),
		cmocka_unit_test_teardown(test_qos_stats_invalid_param_port, teardown),
		cmocka_unit_test_teardown(test_qos_stats_invalid_param_pid, teardown),
	};

	return cmocka_run_group_tests(tests, group_setup, group_teardown);
}