- 'pass': snaps_creds.password,
- 'tenant_n': snaps_creds.project_name,
- 'user_n': snaps_creds.username}
- with open(credentials_yaml, 'w') as yfile:
- yfile.write(CREDS_TEMPLATE2.format(**creds_data))
- cmd = ['juju', 'add-credential', 'abot-epc', '-f', credentials_yaml,
- '--replace']
- output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
- self.__logger.info("%s\n%s", " ".join(cmd), output)
-
- def _register_credentials_v3(self):
- self.__logger.info("Creating Credentials for Abot-epc .....")
- user_creator = self._bypass_juju_netdiscovery_bug(
- 'juju_network_discovery_bug')
- snaps_creds = user_creator.get_os_creds(self.snaps_creds.project_name)
- self.__logger.debug("snaps creds: %s", snaps_creds)
- credentials_yaml = os.path.join(self.res_dir, "credentials.yaml")
- creds_data = {
- 'pass': snaps_creds.password,
- 'tenant_n': snaps_creds.project_name,
- 'user_n': snaps_creds.username,
- 'project_domain_n': snaps_creds.project_domain_name,
- 'user_domain_n': snaps_creds.user_domain_name}
- with open(credentials_yaml, 'w') as yfile:
- yfile.write(CREDS_TEMPLATE3.format(**creds_data))
- cmd = ['juju', 'add-credential', 'abot-epc', '-f', credentials_yaml,
- '--replace']
- output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
- self.__logger.info("%s\n%s", " ".join(cmd), output)
-
- def _add_custom_rule(self, sec_grp_name):
- """ To add custom rule for SCTP Traffic """
-
- security_group = OpenStackSecurityGroup(
- self.snaps_creds,
- SecurityGroupConfig(
- name=sec_grp_name))
-
- security_group.create()
-
- # Add custom security rule to the obtained Security Group
- self.__logger.info("Adding SCTP ingress rule to SG:%s",
- security_group.sec_grp_settings.name)
-
- try:
- security_group.add_rule(SecurityGroupRuleConfig(
- sec_grp_name=sec_grp_name, direction=Direction.ingress,
- protocol=Protocol.sctp))
- except Exception: # pylint: disable=broad-except
- self.__logger.exception(
- "Some issue encountered with adding SCTP security rule ...")
-
- def prepare(self):
- """Prepare testcase (Additional pre-configuration steps)."""
- self.__logger.info("Additional pre-configuration steps")
- super(JujuEpc, self).prepare()
- try:
- os.makedirs(self.res_dir)
- except OSError as ex:
- if ex.errno != errno.EEXIST:
- self.__logger.exception("Cannot create %s", self.res_dir)
- raise vnf.VnfPreparationException
-
- self.__logger.info("ENV:\n%s", env.string())
-
- self.public_auth_url = keystone_utils.get_endpoint(
- self.snaps_creds, 'identity')
-
- # it enforces a versioned public identity endpoint as juju simply
- # adds /auth/tokens wich fails vs an unversioned endpoint.
- if not self.public_auth_url.endswith(('v3', 'v3/', 'v2.0', 'v2.0/')):
- self.public_auth_url = six.moves.urllib.parse.urljoin(
- self.public_auth_url, 'v3')
- self._register_cloud()
- if self.snaps_creds.identity_api_version == 3:
- self._register_credentials_v3()
- else:
- self._register_credentials_v2()
+ 'pass': self.project.password,
+ 'tenant_n': self.project.project.name,
+ 'user_n': self.project.user.name,
+ 'project_domain_n': self.cloud.auth.get(
+ "project_domain_name", "Default"),
+ 'user_domain_n': self.cloud.auth.get(
+ "user_domain_name", "Default")}
+ with open(credentials_yaml, 'w', encoding='utf-8') as yfile:
+ yfile.write(CREDS_TEMPLATE.format(**creds_data))
+ scpc = scp.SCPClient(self.ssh.get_transport())
+ scpc.put(credentials_yaml, remote_path='~/')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju add-credential abot-epc -f credentials.yaml '
+ ' --replace --debug')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def _publish_image(self):
+ region_name = self.cloud.region_name if self.cloud.region_name else (
+ 'RegionOne')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju metadata generate-image -d /home/ubuntu '
+ f'-i {self.image.id} -s xenial -r {region_name} '
+ f'-u {self.public_auth_url}')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def publish_image_alt(self, name=None):
+ image_alt = super().publish_image_alt(name)
+ region_name = self.cloud.region_name if self.cloud.region_name else (
+ 'RegionOne')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju metadata generate-image -d /home/ubuntu '
+ f'-i {image_alt.id} -s trusty -r {region_name} '
+ f'-u {self.public_auth_url}')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return image_alt