149 lines
4.9 KiB
Python
149 lines
4.9 KiB
Python
# Copyright 2018 Creu Blanca
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
|
|
|
|
from unittest.mock import patch
|
|
|
|
from odoo.tests.common import TransactionCase
|
|
from odoo.tools import mute_logger
|
|
|
|
from odoo.addons.nsca_client.models.nsca_server import NscaServer
|
|
|
|
|
|
class Popen:
|
|
def __init__(self, cmd, stdout, stdin, stderr):
|
|
self.cmd = cmd
|
|
self.stdout = stdout
|
|
self.stdin = stdin
|
|
self.stderr = stderr
|
|
|
|
# flake8: noqa: B902
|
|
def communicate(_input):
|
|
return ["test"]
|
|
|
|
|
|
class TestNsca(TransactionCase):
|
|
def test_nsca(self):
|
|
server = self.env["nsca.server"].create(
|
|
{
|
|
"name": "localhost",
|
|
"password": "pass",
|
|
"encryption_method": "3",
|
|
"node_hostname": "odoodev",
|
|
}
|
|
)
|
|
self.assertTrue(server.config_file_path)
|
|
with patch("subprocess.Popen") as post:
|
|
post.return_value = Popen
|
|
check = self.env["nsca.check"].create(
|
|
{
|
|
"server_id": server.id,
|
|
"service": "test",
|
|
"nsca_model": "nsca.server",
|
|
"nsca_function": "current_status",
|
|
}
|
|
)
|
|
self.assertTrue(check.model_id)
|
|
self.env["nsca.check"]._cron_check(check.id)
|
|
|
|
def test_write(self):
|
|
server = self.env["nsca.server"].create(
|
|
{
|
|
"name": "localhost",
|
|
"password": "pass",
|
|
"encryption_method": "3",
|
|
"node_hostname": "odoodev",
|
|
}
|
|
)
|
|
self.assertTrue(server.config_file_path)
|
|
check = self.env["nsca.check"].create(
|
|
{
|
|
"server_id": server.id,
|
|
"service": "test",
|
|
"nsca_model": "nsca.server",
|
|
"nsca_function": "current_status",
|
|
}
|
|
)
|
|
check.cron_id.state = "object_create"
|
|
check.write({"interval_number": 1})
|
|
self.assertEqual(check.cron_id.state, "object_create")
|
|
check.write({"service": "change"})
|
|
self.assertNotEqual(check.cron_id.state, "object_create")
|
|
|
|
def test_void_failure(self):
|
|
server = self.env["nsca.server"].create(
|
|
{
|
|
"name": "localhost",
|
|
"password": "pass",
|
|
"encryption_method": "3",
|
|
"node_hostname": "odoodev",
|
|
}
|
|
)
|
|
check = self.env["nsca.check"].create(
|
|
{
|
|
"server_id": server.id,
|
|
"service": "test",
|
|
"nsca_model": "nsca.server",
|
|
"allow_void_result": False,
|
|
"nsca_function": "_check_send_nsca_command",
|
|
}
|
|
)
|
|
with patch("subprocess.Popen") as post:
|
|
post.return_value = Popen
|
|
with mute_logger("odoo.addons.nsca_client.models.nsca_check"):
|
|
self.env["nsca.check"]._cron_check(check.id)
|
|
post.assert_called_once()
|
|
|
|
def test_void_ok(self):
|
|
server = self.env["nsca.server"].create(
|
|
{
|
|
"name": "localhost",
|
|
"password": "pass",
|
|
"encryption_method": "3",
|
|
"node_hostname": "odoodev",
|
|
}
|
|
)
|
|
self.assertEqual(server.check_count, 0)
|
|
check = self.env["nsca.check"].create(
|
|
{
|
|
"server_id": server.id,
|
|
"service": "test",
|
|
"nsca_model": "nsca.server",
|
|
"allow_void_result": True,
|
|
"nsca_function": "_check_send_nsca_command",
|
|
}
|
|
)
|
|
self.assertEqual(server.check_count, 1)
|
|
action = server.show_checks()
|
|
self.assertEqual(check, self.env["nsca.check"].browse(action["res_id"]))
|
|
self.assertEqual(check, self.env["nsca.check"].search(action["domain"]))
|
|
with patch("subprocess.Popen") as post:
|
|
post.return_value = Popen
|
|
self.env["nsca.check"]._cron_check(check.id)
|
|
post.assert_not_called()
|
|
|
|
def test_values(self):
|
|
server = self.env["nsca.server"].create(
|
|
{
|
|
"name": "localhost",
|
|
"password": "pass",
|
|
"encryption_method": "3",
|
|
"node_hostname": "odoodev",
|
|
}
|
|
)
|
|
check = self.env["nsca.check"].create(
|
|
{
|
|
"server_id": server.id,
|
|
"service": "test",
|
|
"nsca_model": "nsca.server",
|
|
"allow_void_result": False,
|
|
"nsca_function": "_check_send_nsca_command",
|
|
}
|
|
)
|
|
with patch("subprocess.Popen") as post:
|
|
post.return_value = Popen
|
|
with patch.object(NscaServer, "_check_send_nsca_command") as func:
|
|
func.return_value = ("OK", "RESULT")
|
|
self.env["nsca.check"]._cron_check(check.id)
|
|
func.assert_called()
|
|
post.assert_called_once()
|