mirror of
https://github.com/ansible/awx.git
synced 2026-01-13 11:00:03 -03:30
Ported old test_licences.py tests
This commit is contained in:
Akita Noek
parent
096a0a262a
commit
66f86de3fa
132
awx/main/tests/functional/core/test_licenses.py
Normal file
132
awx/main/tests/functional/core/test_licenses.py
Normal file
@ -0,0 +1,132 @@
|
||||
# Copyright (c) 2015 Ansible, Inc.
|
||||
# All Rights Reserved.
|
||||
|
||||
import json
|
||||
import mock
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
|
||||
from awx.main.models import Host
|
||||
from awx.main.task_engine import TaskSerializer, TaskEngager
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_license_writer(inventory, admin):
|
||||
writer = TaskEngager(
|
||||
company_name='acmecorp',
|
||||
contact_name='Michael DeHaan',
|
||||
contact_email='michael@ansibleworks.com',
|
||||
license_date=25000, # seconds since epoch
|
||||
instance_count=500)
|
||||
|
||||
data = writer.get_data()
|
||||
|
||||
Host.objects.bulk_create(
|
||||
[
|
||||
Host(
|
||||
name='host.%d' % n,
|
||||
inventory=inventory,
|
||||
created_by=admin,
|
||||
modified=datetime.now(),
|
||||
created=datetime.now())
|
||||
for n in range(12)
|
||||
]
|
||||
)
|
||||
|
||||
assert data['instance_count'] == 500
|
||||
assert data['contact_name'] == 'Michael DeHaan'
|
||||
assert data['contact_email'] == 'michael@ansibleworks.com'
|
||||
assert data['license_date'] == 25000
|
||||
assert data['license_key'] == "11bae31f31c6a6cdcb483a278cdbe98bd8ac5761acd7163a50090b0f098b3a13"
|
||||
|
||||
strdata = writer.get_string()
|
||||
strdata_loaded = json.loads(strdata)
|
||||
assert strdata_loaded == data
|
||||
|
||||
reader = TaskSerializer()
|
||||
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['available_instances'] == 500
|
||||
assert vdata['current_instances'] == 12
|
||||
assert vdata['free_instances'] == 488
|
||||
assert vdata['date_warning'] is True
|
||||
assert vdata['date_expired'] is True
|
||||
assert vdata['license_date'] == 25000
|
||||
assert vdata['time_remaining'] < 0
|
||||
assert vdata['valid_key'] is True
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['subscription_name']
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_expired_licenses():
|
||||
reader = TaskSerializer()
|
||||
writer = TaskEngager(
|
||||
company_name='Tower',
|
||||
contact_name='Tower Admin',
|
||||
contact_email='tower@ansible.com',
|
||||
license_date=int(time.time() - 3600),
|
||||
instance_count=100,
|
||||
trial=True)
|
||||
strdata = writer.get_string()
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['grace_period_remaining'] < 0
|
||||
|
||||
writer = TaskEngager(
|
||||
company_name='Tower',
|
||||
contact_name='Tower Admin',
|
||||
contact_email='tower@ansible.com',
|
||||
license_date=int(time.time() - 2592001),
|
||||
instance_count=100,
|
||||
trial=False)
|
||||
strdata = writer.get_string()
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['grace_period_remaining'] < 0
|
||||
|
||||
writer = TaskEngager(
|
||||
company_name='Tower',
|
||||
contact_name='Tower Admin',
|
||||
contact_email='tower@ansible.com',
|
||||
license_date=int(time.time() - 3600),
|
||||
instance_count=100,
|
||||
trial=False)
|
||||
strdata = writer.get_string()
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['grace_period_remaining'] > 0
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_aws_license():
|
||||
os.environ['AWX_LICENSE_FILE'] = 'non-existent-license-file.json'
|
||||
|
||||
h, path = tempfile.mkstemp()
|
||||
with os.fdopen(h, 'w') as f:
|
||||
json.dump({'instance_count': 100}, f)
|
||||
|
||||
def fetch_ami(_self):
|
||||
_self.attributes['ami-id'] = 'ami-00000000'
|
||||
return True
|
||||
|
||||
def fetch_instance(_self):
|
||||
_self.attributes['instance-id'] = 'i-00000000'
|
||||
return True
|
||||
|
||||
with mock.patch('awx.main.task_engine.TEMPORARY_TASK_FILE', path):
|
||||
with mock.patch('awx.main.task_engine.TemporaryTaskEngine.fetch_ami', fetch_ami):
|
||||
with mock.patch('awx.main.task_engine.TemporaryTaskEngine.fetch_instance', fetch_instance):
|
||||
reader = TaskSerializer()
|
||||
license = reader.from_file()
|
||||
assert license['is_aws']
|
||||
assert license['time_remaining']
|
||||
assert license['free_instances'] > 0
|
||||
assert license['grace_period_remaining'] > 0
|
||||
|
||||
os.unlink(path)
|
||||
@ -1,147 +0,0 @@
|
||||
# Copyright (c) 2015 Ansible, Inc.
|
||||
# All Rights Reserved.
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from awx.main.models import Host, Inventory, Organization
|
||||
from awx.main.tests.base import BaseTest
|
||||
import awx.main.task_engine
|
||||
from awx.main.task_engine import * # noqa
|
||||
|
||||
class LicenseTests(BaseTest):
|
||||
|
||||
def setUp(self):
|
||||
self.start_redis()
|
||||
self.setup_instances()
|
||||
super(LicenseTests, self).setUp()
|
||||
self.setup_users()
|
||||
u = self.super_django_user
|
||||
org = Organization.objects.create(name='o1', created_by=u)
|
||||
org.admin_role.members.add(self.normal_django_user)
|
||||
self.inventory = Inventory.objects.create(name='hi', organization=org, created_by=u)
|
||||
Host.objects.create(name='a1', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a2', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a3', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a4', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a5', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a6', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a7', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a8', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a9', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a10', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a11', inventory=self.inventory, created_by=u)
|
||||
Host.objects.create(name='a12', inventory=self.inventory, created_by=u)
|
||||
self._temp_task_file = awx.main.task_engine.TEMPORARY_TASK_FILE
|
||||
self._temp_task_fetch_ami = awx.main.task_engine.TemporaryTaskEngine.fetch_ami
|
||||
self._temp_task_fetch_instance = awx.main.task_engine.TemporaryTaskEngine.fetch_instance
|
||||
|
||||
def tearDown(self):
|
||||
awx.main.task_engine.TEMPORARY_TASK_FILE = self._temp_task_file
|
||||
awx.main.task_engine.TemporaryTaskEngine.fetch_ami = self._temp_task_fetch_ami
|
||||
awx.main.task_engine.TemporaryTaskEngine.fetch_instance = self._temp_task_fetch_instance
|
||||
super(LicenseTests, self).tearDown()
|
||||
self.stop_redis()
|
||||
|
||||
def test_license_writer(self):
|
||||
|
||||
writer = TaskEngager(
|
||||
company_name='acmecorp',
|
||||
contact_name='Michael DeHaan',
|
||||
contact_email='michael@ansibleworks.com',
|
||||
license_date=25000, # seconds since epoch
|
||||
instance_count=500)
|
||||
|
||||
data = writer.get_data()
|
||||
|
||||
assert data['instance_count'] == 500
|
||||
assert data['contact_name'] == 'Michael DeHaan'
|
||||
assert data['contact_email'] == 'michael@ansibleworks.com'
|
||||
assert data['license_date'] == 25000
|
||||
assert data['license_key'] == "11bae31f31c6a6cdcb483a278cdbe98bd8ac5761acd7163a50090b0f098b3a13"
|
||||
|
||||
strdata = writer.get_string()
|
||||
strdata_loaded = json.loads(strdata)
|
||||
assert strdata_loaded == data
|
||||
|
||||
reader = TaskSerializer()
|
||||
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['available_instances'] == 500
|
||||
assert vdata['current_instances'] == 12
|
||||
assert vdata['free_instances'] == 488
|
||||
assert vdata['date_warning'] is True
|
||||
assert vdata['date_expired'] is True
|
||||
assert vdata['license_date'] == 25000
|
||||
assert vdata['time_remaining'] < 0
|
||||
assert vdata['valid_key'] is True
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['subscription_name']
|
||||
|
||||
def test_expired_licenses(self):
|
||||
reader = TaskSerializer()
|
||||
writer = TaskEngager(
|
||||
company_name='Tower',
|
||||
contact_name='Tower Admin',
|
||||
contact_email='tower@ansible.com',
|
||||
license_date=int(time.time() - 3600),
|
||||
instance_count=100,
|
||||
trial=True)
|
||||
strdata = writer.get_string()
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['grace_period_remaining'] < 0
|
||||
|
||||
writer = TaskEngager(
|
||||
company_name='Tower',
|
||||
contact_name='Tower Admin',
|
||||
contact_email='tower@ansible.com',
|
||||
license_date=int(time.time() - 2592001),
|
||||
instance_count=100,
|
||||
trial=False)
|
||||
strdata = writer.get_string()
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['grace_period_remaining'] < 0
|
||||
|
||||
writer = TaskEngager(
|
||||
company_name='Tower',
|
||||
contact_name='Tower Admin',
|
||||
contact_email='tower@ansible.com',
|
||||
license_date=int(time.time() - 3600),
|
||||
instance_count=100,
|
||||
trial=False)
|
||||
strdata = writer.get_string()
|
||||
vdata = reader.from_string(strdata)
|
||||
|
||||
assert vdata['compliant'] is False
|
||||
assert vdata['grace_period_remaining'] > 0
|
||||
|
||||
def test_aws_license(self):
|
||||
os.environ['AWX_LICENSE_FILE'] = 'non-existent-license-file.json'
|
||||
h, path = tempfile.mkstemp()
|
||||
self._temp_paths.append(path)
|
||||
with os.fdopen(h, 'w') as f:
|
||||
json.dump({'instance_count': 100}, f)
|
||||
awx.main.task_engine.TEMPORARY_TASK_FILE = path
|
||||
|
||||
def fetch_ami(_self):
|
||||
_self.attributes['ami-id'] = 'ami-00000000'
|
||||
return True
|
||||
|
||||
def fetch_instance(_self):
|
||||
_self.attributes['instance-id'] = 'i-00000000'
|
||||
return True
|
||||
|
||||
awx.main.task_engine.TemporaryTaskEngine.fetch_ami = fetch_ami
|
||||
awx.main.task_engine.TemporaryTaskEngine.fetch_instance = fetch_instance
|
||||
reader = TaskSerializer()
|
||||
license = reader.from_file()
|
||||
self.assertTrue(license['is_aws'])
|
||||
self.assertTrue(license['time_remaining'])
|
||||
self.assertTrue(license['free_instances'] > 0)
|
||||
self.assertTrue(license['grace_period_remaining'] > 0)
|
||||
Loading…
x
Reference in New Issue
Block a user