# Copyright © 2022 Aravinth Manivannan # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # 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 Affero 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 . import time from io import StringIO from django.test import TestCase, Client, override_settings from django.core.management import call_command from django.core import mail from dash.models import Instance, InstanceConfiguration from accounts.tests import register_util, login_util from dash.tests import create_configurations, create_instance_util, infra_custom_config from infrastructure.management.commands.vm import translate_sizes from billing.utils import payment_fullfilled from .utils import Infra, Worker, create_vm_if_not_exists, delete_vm from .models import InstanceCreated, Job, JobType class InfraUtilTest(TestCase): """ Tests billing system """ def setUp(self): self.username = "infrautil_user" register_util(t=self, username=self.username) self.user.is_staff = True self.user.save() create_configurations(t=self) # @override_settings(HOSTEA=infra_custom_config(test_name="test_path_util")) # def test_path_utils(self): # infra = Infra() # subdomain = "foo" # base = infra.repo_path # # self.assertEqual( # base.joinpath(f"inventory/host_vars/{subdomain}-host/"), # infra._host_vars_dir(subdomain=subdomain), # ) # # self.assertEqual( # base.joinpath(f"inventory/host_vars/{subdomain}-host/gitea.yml"), # infra._gitea_path(subdomain=subdomain), # ) # # self.assertEqual( # base.joinpath(f"inventory/host_vars/{subdomain}-host/provision.yml"), # infra._provision_path(subdomain=subdomain), # ) # # self.assertEqual( # base.joinpath(f"inventory/{subdomain}-backup.yml"), # infra._backup_path(subdomain=subdomain), # ) # # self.assertEqual( # base.joinpath(f"hosts-scripts/{subdomain}-host.sh"), # infra._hostscript_path(subdomain=subdomain), # ) # # @override_settings(HOSTEA=infra_custom_config(test_name="test_add_vm")) # def test_add_vm(self): # c = Client() # login_util(self, c, "accounts.home") # subdomain = "add_vm" # # create_instance_util( # t=self, c=c, instance_name=subdomain, config=self.instance_config[0] # ) # # instance = Instance.objects.get(name=subdomain) # # infra = Infra() # before_add = infra._sha() # (password, after_add) = infra.add_vm(instance=instance) # self.assertNotEqual(before_add, after_add) # # before_rm = after_add # after_rm = infra.remove_vm(instance=instance) # self.assertNotEqual(before_rm, after_rm) # # @override_settings(HOSTEA=infra_custom_config(test_name="test_cmd")) # def test_cmd(self): # subdomain = "cmd_vm" # infra = Infra() # c = Client() # login_util(self, c, "accounts.home") # # self.assertEqual(Instance.objects.filter(name=subdomain).exists(), False) # # username exists # call_command( # "vm", "create", subdomain, f"--owner={self.username}", "--flavor=medium" # ) # # instance = Instance.objects.get(name=subdomain) # # self.assertEqual(infra.get_flavor(instance=instance), "openstack_flavor_medium") # # self.assertEqual(instance.owned_by, self.user) # self.assertEqual( # instance.configuration_id, InstanceConfiguration.objects.get(name="s1-4") # ) # # instance_created = InstanceCreated.objects.get(instance=instance) # self.assertEqual(instance_created.instance, instance) # # self.assertEqual(instance_created.created, True) # # # run create vm command again with same configuration to crudely check idempotency # call_command( # "vm", "create", subdomain, f"--owner={self.username}", "--flavor=medium" # ) # # # run create vm command again with different configuration but same name # # to crudely check idempotency # call_command( # "vm", "create", subdomain, f"--owner={self.username}", "--flavor=large" # ) # instance.refresh_from_db() # # verify new size is updated in DB # self.assertEqual( # str.strip(instance.configuration_id.name) # == str.strip(translate_sizes("large")), # True, # ) # # # verify new size is updated in repository # self.assertEqual( # str.strip(infra.translate_size(instance=instance)) # == str.strip(infra.get_flavor(instance=instance)), # True, # ) # # call_command("vm", "delete", subdomain) # # self.assertEqual(Instance.objects.filter(name=subdomain).exists(), False) # host_vars_dir = infra._host_vars_dir(subdomain) # self.assertEqual(host_vars_dir.exists(), False) # # # run delete VM command to crudely check idempotency # call_command("vm", "delete", subdomain) # # def test_worker(self): # subdomain = "gitea" # yes, gitea.hostea.org exists. will use it till I # # figure out how to use requests_mock within django # c = Client() # login_util(self, c, "accounts.home") # create_instance_util( # t=self, c=c, instance_name=subdomain, config=self.instance_config[0] # ) # # instance = Instance.objects.get(name=subdomain) # job = Job.objects.create(instance=instance, job_type=JobType.PING) # gitea_uri = Infra.get_gitea_uri(instance=instance) # print(f"mocking {gitea_uri}") # # w = Worker(job=job) # w.start() # time.sleep(15) # self.assertEqual(w.is_alive(), False) # w.join() # self.assertEqual( # Job.objects.filter(instance=instance, job_type=JobType.PING).exists(), True # ) # @override_settings(HOSTEA=infra_custom_config(test_name="test_vm_delete_payments")) def test_vm_delete_payments(self): """ Test if the dashboard generates invoices for a VM crated with a name matching a VM name that was deleted that existed. ref: https://gitea.hostea.org/Hostea/dashboard/issues/38#issuecomment-1162 """ c = Client() login_util(self, c, "accounts.home") instance_name = "trmpayments" infra = Infra() create_instance_util( t=self, c=c, instance_name=instance_name, config=self.instance_config[0] ) instance = Instance.objects.get(name=instance_name) self.assertEqual(payment_fullfilled(instance=instance), True) create_vm_if_not_exists(instance=instance) # delete VM delete_vm(instance=instance) self.assertEqual(Instance.objects.filter(name=instance_name).exists(), False) # re-create VM with management command as it bypasses payments. We # usually use create_instance_util but it will pay for the instance too call_command( "vm", "create", instance_name, f"--owner={self.username}", "--flavor=medium" ) # verify VM is created self.assertEqual(Instance.objects.filter(name=instance_name).exists(), True) # verify payment is unfulfilled instance = Instance.objects.get(name=instance_name) self.assertEqual(payment_fullfilled(instance=instance), False) # generate invoice stdout = StringIO() stderr = StringIO() call_command( "generate_invoice", stdout=stdout, stderr=stderr, ) out = stdout.getvalue() print("out") print(out) self.assertEqual(instance_name in out, True) self.assertEqual(f"Payment not fulfilled for instance: {instance}" in out, True) staff_notification = None for m in mail.outbox: if "New instance alert" in m.subject: staff_notification = m break self.assertEqual(staff_notification.to[0], self.email) self.assertEqual( "[Gna!] New instance alert" in staff_notification.subject, True ) self.assertEqual( "A customer has purchased a new instance. Please find the details below:" in staff_notification.body, True, )