Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -3,22 +3,18 @@
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest

mysql_location = None
postgres_location = None
test_location = None
resource_random_name = None


def pytest_addoption(parser):
parser.addoption("--mysql-location", action="store", default="eastus2euap")
parser.addoption("--postgres-location", action="store", default="eastus2euap")
parser.addoption("--test-location", action="store", default="eastus2euap")
parser.addoption("--resource-random-name", action="store", default="clirecording")


def pytest_configure(config):
global mysql_location, postgres_location, resource_random_name # pylint:disable=global-statement
mysql_location = config.getoption('--mysql-location')
postgres_location = config.getoption('--postgres-location')
global test_location, resource_random_name # pylint:disable=global-statement
test_location = config.getoption('--test-location')
resource_random_name = config.getoption('--resource-random-name')


Expand All @@ -29,6 +25,7 @@ def pytest_configure(config):
PROXY_SERVER_FILE = './proxy_server_cache.txt'
IOPS_SERVER_FILE = './iops_server_cache.txt'
REPLICA_SERVER_FILE = './replica_server_cache.txt'
SINGLE_AVAILABILITY_FILE = './single_availability.txt'

def skip_if_test_failed(filename):
with open(filename, "r") as f:
Expand All @@ -37,6 +34,14 @@ def skip_if_test_failed(filename):
if result == 'FAIL':
pytest.skip("skipping the test due to dependent resource provision failure")

@pytest.fixture()
def single_availability_zone_check():
with open(SINGLE_AVAILABILITY_FILE, "r") as f:
result = f.readline()

if result == "TRUE":
pytest.skip("skipping the test due to non supported feature in single availability zone")

@pytest.fixture()
def regular_server_provision_check():
skip_if_test_failed(REGULAR_SERVER_FILE)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,18 +22,18 @@
from .test_rdbms_flexible_commands import (
ServerPreparer
)
from .conftest import mysql_location
from .conftest import test_location

if mysql_location is None:
mysql_location = 'eastus2euap'
if test_location is None:
test_location = 'eastus2euap'


# Local context test is separated out from the rest of the test due to daily pipeline run issue
class MySqlFlexibleServerLocalContextScenarioTest(FlexibleServerLocalContextScenarioTest):

mysql_location = mysql_location
test_location = test_location

@AllowLargeResponse()
@ResourceGroupPreparer(location=mysql_location)
@ResourceGroupPreparer(location=test_location)
def test_mysql_flexible_server_local_context(self, resource_group):
self._test_flexible_server_local_context('mysql', resource_group)
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,8 @@
write_failed_result,
write_succeeded_result
)
from .conftest import mysql_location, REGULAR_SERVER_FILE, VNET_SERVER_FILE, VNET_HA_SERVER_FILE, HA_SERVER_FILE, PROXY_SERVER_FILE, IOPS_SERVER_FILE, REPLICA_SERVER_FILE
from .conftest import test_location, SINGLE_AVAILABILITY_FILE, REGULAR_SERVER_FILE, VNET_SERVER_FILE, VNET_HA_SERVER_FILE, HA_SERVER_FILE, PROXY_SERVER_FILE, IOPS_SERVER_FILE, REPLICA_SERVER_FILE
from ..._flexible_server_util import get_mysql_list_skus_info

SERVER_NAME_PREFIX = 'clitest-'
RG_NAME_PREFIX = 'clitest.rg'
Expand All @@ -44,8 +45,8 @@
SOURCE_VNET_SERVER_PREFIX = 'clitest-server-vnet-mysql-'
SOURCE_VNET_HA_SERVER_PREFIX = 'clitest-server-vnet-ha-mysql-'

if mysql_location is None:
mysql_location = 'eastus2euap'
if test_location is None:
test_location = 'eastus2euap'


class MySqlFlexibleServerRegularMgmtScenarioTest(FlexibleServerRegularMgmtScenarioTest):
Expand All @@ -57,11 +58,18 @@ def __init__(self, method_name):
self.server2 = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'diff-tier1')
self.server3 = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'diff-tier2')
self.restore_server = 'restore-' + self.server[:55]
self.location = mysql_location
self.location = test_location
_, single_az = get_mysql_list_skus_info(self, test_location)
with open(SINGLE_AVAILABILITY_FILE, "w") as f:
if single_az:
f.write("TRUE")
else:
f.write("FALSE")


@pytest.mark.order(1)
def test_mysql_flexible_server_mgmt_prepare(self):
self.cmd('az group create --location {} --name {}'.format(mysql_location, self.resource_group))
self.cmd('az group create --location {} --name {}'.format(test_location, self.resource_group))

@AllowLargeResponse()
@pytest.mark.order(2)
Expand Down Expand Up @@ -228,12 +236,12 @@ def __init__(self, method_name):
super(MySqlFlexibleServerIopsMgmtScenarioTest, self).__init__(method_name)
self.resource_group = self.create_random_name(RG_NAME_PREFIX, RG_NAME_MAX_LENGTH)
self.server = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'iops')
self.location = mysql_location
self.location = test_location

@AllowLargeResponse()
@pytest.mark.order(1)
def test_mysql_flexible_server_iops_prepare(self):
self.cmd('az group create --location {} --name {}'.format(mysql_location, self.resource_group))
self.cmd('az group create --location {} --name {}'.format(test_location, self.resource_group))

@AllowLargeResponse()
@pytest.mark.order(2)
Expand Down Expand Up @@ -280,17 +288,23 @@ class MySqlFlexibleServerVnetServerMgmtScenarioTest(FlexibleServerVnetServerMgmt

def __init__(self, method_name):
super(MySqlFlexibleServerVnetServerMgmtScenarioTest, self).__init__(method_name)
self.location = mysql_location
self.location = test_location
self.resource_group = self.create_random_name(RG_NAME_PREFIX, RG_NAME_MAX_LENGTH)
self.server = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'vnet')
self.server2 = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'vnet-ha')
self.restore_server = 'restore-' + self.server[:55]
self.restore_server2 = 'restore-' + self.server2[:55]
self.current_time = datetime.utcnow()
_, single_az = get_mysql_list_skus_info(self, test_location)
with open(SINGLE_AVAILABILITY_FILE, "w") as f:
if single_az:
f.write("TRUE")
else:
f.write("FALSE")

@pytest.mark.order(1)
def test_mysql_flexible_server_vnet_server_prepare(self):
self.cmd('az group create --location {} --name {}'.format(mysql_location, self.resource_group))
self.cmd('az group create --location {} --name {}'.format(test_location, self.resource_group))

@AllowLargeResponse()
@pytest.mark.order(2)
Expand All @@ -307,6 +321,13 @@ def test_mysql_flexible_server_vnet_server_create(self):
@pytest.mark.order(3)
@pytest.mark.execution_timeout(3600)
def test_mysql_flexible_server_vnet_ha_server_create(self):
with open(SINGLE_AVAILABILITY_FILE, "r") as f:
result = f.readline()

if result == "TRUE":
write_failed_result(VNET_HA_SERVER_FILE)
pytest.skip("skipping the test due to non supported feature in single availability zone")

try:
self._test_flexible_server_vnet_ha_server_create('mysql', self.resource_group, self.server2)
write_succeeded_result(VNET_HA_SERVER_FILE)
Expand All @@ -323,6 +344,7 @@ def test_mysql_flexible_server_vnet_server_update_scale_up(self):

@AllowLargeResponse()
@pytest.mark.order(5)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.usefixtures("vnet_ha_server_provision_check")
def test_mysql_flexible_server_vnet_ha_server_delete(self):
self._test_flexible_server_vnet_server_delete('mysql', self.resource_group, self.server2)
Expand All @@ -339,16 +361,17 @@ def test_mysql_flexible_server_vnet_server_delete(self):
def test_mysql_flexible_server_vnet_server_restore(self):
self._test_flexible_server_vnet_server_restore('mysql', SOURCE_RG, SOURCE_VNET_SERVER_PREFIX + self.location, self.restore_server)

# @AllowLargeResponse()
# @pytest.mark.order(8)
# @pytest.mark.execution_timeout(5400)
# def test_mysql_flexible_server_vnet_ha_server_restore(self):
# self._test_flexible_server_vnet_server_restore('mysql', SOURCE_RG, SOURCE_VNET_HA_SERVER_PREFIX + self.location, self.restore_server2)
@AllowLargeResponse()
@pytest.mark.order(8)
@pytest.mark.execution_timeout(5400)
@pytest.mark.usefixtures("single_availability_zone_check")
def test_mysql_flexible_server_vnet_ha_server_restore(self):
self._test_flexible_server_vnet_server_restore('mysql', SOURCE_RG, SOURCE_VNET_HA_SERVER_PREFIX + self.location, self.restore_server2)


class MySqlFlexibleServerProxyResourceMgmtScenarioTest(FlexibleServerProxyResourceMgmtScenarioTest):

mysql_location = mysql_location
test_location = test_location

def __init__(self, method_name):
super(MySqlFlexibleServerProxyResourceMgmtScenarioTest, self).__init__(method_name)
Expand All @@ -360,8 +383,8 @@ def __init__(self, method_name):
@pytest.mark.execution_timeout(3600)
def test_mysql_flexible_server_proxy_resource_mgmt_prepare(self):
try:
self.cmd('az group create --location {} --name {}'.format(mysql_location, self.resource_group))
self.cmd('az {} flexible-server create -l {} -g {} -n {} --public-access none'.format('mysql', mysql_location, self.resource_group, self.server))
self.cmd('az group create --location {} --name {}'.format(test_location, self.resource_group))
self.cmd('az {} flexible-server create -l {} -g {} -n {} --public-access none'.format('mysql', test_location, self.resource_group, self.server))
write_succeeded_result(PROXY_SERVER_FILE)
except:
write_failed_result(PROXY_SERVER_FILE)
Expand Down Expand Up @@ -394,19 +417,33 @@ class MySqlFlexibleServerHighAvailabilityMgmt(FlexibleServerHighAvailabilityMgmt
def __init__(self, method_name):
super(MySqlFlexibleServerHighAvailabilityMgmt, self).__init__(method_name)
self.current_time = datetime.utcnow()
self.location = mysql_location
self.location = test_location
self.resource_group = self.create_random_name(RG_NAME_PREFIX, RG_NAME_MAX_LENGTH)
self.server = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'ha')
self.restore_server = 'restore-' + self.server[:55]
_, single_az = get_mysql_list_skus_info(self, test_location)
with open(SINGLE_AVAILABILITY_FILE, "w") as f:
if single_az:
f.write("TRUE")
else:
f.write("FALSE")

@pytest.mark.order(1)
@pytest.mark.usefixtures("single_availability_zone_check")
def test_mysql_flexible_server_high_availability_prepare(self):
self.cmd('az group create --location {} --name {}'.format(mysql_location, self.resource_group))
self.cmd('az group create --location {} --name {}'.format(test_location, self.resource_group))

@AllowLargeResponse()
@pytest.mark.order(2)
@pytest.mark.execution_timeout(3600)
def test_mysql_flexible_server_high_availability_create(self):
with open(SINGLE_AVAILABILITY_FILE, "r") as f:
result = f.readline()

if result == "TRUE":
write_failed_result(HA_SERVER_FILE)
pytest.skip("skipping the test due to non supported feature in single availability zone")

try:
self._test_flexible_server_high_availability_create('mysql', self.resource_group, self.server)
write_succeeded_result(HA_SERVER_FILE)
Expand All @@ -417,6 +454,7 @@ def test_mysql_flexible_server_high_availability_create(self):
@AllowLargeResponse()
@pytest.mark.order(3)
@pytest.mark.execution_timeout(3600)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.usefixtures("ha_server_provision_check")
def test_mysql_flexible_server_high_availability_update_parameter(self):
try:
Expand All @@ -429,6 +467,7 @@ def test_mysql_flexible_server_high_availability_update_parameter(self):
@AllowLargeResponse()
@pytest.mark.order(4)
@pytest.mark.execution_timeout(3600)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.usefixtures("ha_server_provision_check")
def test_mysql_flexible_server_high_availability_restart(self):
try:
Expand All @@ -441,6 +480,7 @@ def test_mysql_flexible_server_high_availability_restart(self):
@AllowLargeResponse()
@pytest.mark.order(5)
@pytest.mark.execution_timeout(3600)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.usefixtures("ha_server_provision_check")
def test_mysql_flexible_server_high_availability_stop(self):
try:
Expand All @@ -453,6 +493,7 @@ def test_mysql_flexible_server_high_availability_stop(self):
@AllowLargeResponse()
@pytest.mark.order(6)
@pytest.mark.execution_timeout(3600)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.usefixtures("ha_server_provision_check")
def test_mysql_flexible_server_high_availability_start(self):
try:
Expand All @@ -464,6 +505,7 @@ def test_mysql_flexible_server_high_availability_start(self):

@AllowLargeResponse()
@pytest.mark.order(7)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.usefixtures("ha_server_provision_check")
def test_mysql_flexible_server_high_availability_delete(self):
try:
Expand All @@ -475,42 +517,43 @@ def test_mysql_flexible_server_high_availability_delete(self):

@AllowLargeResponse()
@pytest.mark.order(8)
@pytest.mark.usefixtures("single_availability_zone_check")
@pytest.mark.execution_timeout(5400)
def test_mysql_flexible_server_high_availability_restore(self):
self._test_flexible_server_high_availability_restore('mysql', SOURCE_RG, SOURCE_HA_SERVER_PREFIX + self.location, self.restore_server)


class MySqlFlexibleServerValidatorScenarioTest(FlexibleServerValidatorScenarioTest):

mysql_location = mysql_location
test_location = test_location

@AllowLargeResponse()
@ResourceGroupPreparer(location=mysql_location)
@ResourceGroupPreparer(location=test_location)
def test_mysql_flexible_server_mgmt_validator(self, resource_group):
self._test_mgmt_validator('mysql', resource_group)


class MySqlFlexibleServerReplicationMgmtScenarioTest(FlexibleServerReplicationMgmtScenarioTest): # pylint: disable=too-few-public-methods

mysql_location = mysql_location
test_location = test_location

def __init__(self, method_name):
super(MySqlFlexibleServerReplicationMgmtScenarioTest, self).__init__(method_name)
self.resource_group = self.create_random_name(RG_NAME_PREFIX, RG_NAME_MAX_LENGTH)
self.master_server = self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'replica-source')
self.replicas = [self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'replica1'),
self.create_random_name(SERVER_NAME_PREFIX, SERVER_NAME_MAX_LENGTH, 'replica2')]
self.location = mysql_location
self.location = test_location
self.result = None

@AllowLargeResponse()
@pytest.mark.order(1)
@pytest.mark.execution_timeout(5400)
def test_mysql_flexible_server_replica_prepare(self):
try:
self.cmd('az group create --location {} --name {}'.format(mysql_location, self.resource_group))
self.cmd('az group create --location {} --name {}'.format(test_location, self.resource_group))
self.cmd('{} flexible-server create -g {} --name {} -l {} --storage-size {} --public-access none'
.format('mysql', self.resource_group, self.master_server, mysql_location, 256))
.format('mysql', self.resource_group, self.master_server, test_location, 256))
write_succeeded_result(REPLICA_SERVER_FILE)
except:
write_failed_result(REPLICA_SERVER_FILE)
Expand Down Expand Up @@ -573,7 +616,7 @@ def test_mysql_flexible_server_replica_delete(self):

class MySqlFlexibleServerVnetProvisionScenarioTest(FlexibleServerVnetProvisionScenarioTest):

mysql_location = mysql_location
test_location = test_location

@AllowLargeResponse()
@pytest.mark.execution_timeout(3600)
Expand All @@ -594,10 +637,10 @@ def test_mysql_flexible_server_vnet_provision_create_without_parameters(self):

class MySqlFlexibleServerPublicAccessMgmtScenarioTest(FlexibleServerPublicAccessMgmtScenarioTest):

mysql_location = mysql_location
test_location = test_location

@AllowLargeResponse()
@ResourceGroupPreparer(location=mysql_location)
@ResourceGroupPreparer(location=test_location)
@live_only()
@pytest.mark.execution_timeout(5000)
def test_mysql_flexible_server_public_access_mgmt(self, resource_group):
Expand Down
Loading