KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _domain_name method in localstack

Best Python code snippet using localstack_python

digitalocean.py

Source:digitalocean.py

1"""2Python 3 API for accessing DigitalOcean API to synchronise the actual IP3address of a host with its domain name A record.4License Notice5--------------6Copyright 2018 Amanda Wee7Licensed under the Apache License, Version 2.0 (the "License");8you may not use this file except in compliance with the License.9You may obtain a copy of the License at:10http://www.apache.org/licenses/LICENSE-2.011Unless required by applicable law or agreed to in writing, software12distributed under the License is distributed on an "AS IS" BASIS,13WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14See the License for the specific language governing permissions and15limitations under the License.16"""17import requests18from .base import DomainRecordApi, DomainRecordError19class DigitalOceanApi:20    """21    Simple wrapper class for GET, POST, and PUT requests to be made to the22    DigitalOcean API.23    """24    BASE_URI = 'https://api.digitalocean.com/v2/'25    def __init__(self, token):26        self._authorization = 'Bearer {}'.format(token)27    def get(self, path):28        return requests.get(''.join([self.BASE_URI, path]),29                            headers={'Authorization': self._authorization})30    def post(self, path, json_data):31        return requests.post(''.join([self.BASE_URI, path]),32                             json=json_data,33                             headers={'Authorization': self._authorization,34                                      'Content-Type': 'application/json'})35    def put(self, path, json_data):36        return requests.put(''.join([self.BASE_URI, path]),37                            json=json_data,38                            headers={'Authorization': self._authorization,39                                     'Content-Type': 'application/json'})40    def follow_link(self, link):41        return requests.get(link,42                            headers={'Authorization': self._authorization})43@DomainRecordApi.register('digitalocean')44class DigitalOceanDomainRecordApi(DomainRecordApi):45    """46    Domain record API for synchronising the actual IP address of a host with47    its domain name A record on DigitalOcean.48    """49    def __init__(self, domain_name, token):50        if not domain_name:51            raise DomainRecordError('domain name not specified')52        self._domain_name = domain_name53        if not token:54            raise DomainRecordError('DigitalOcean API token not specified')55        self._api = DigitalOceanApi(token)56    def create(self, host_name, actual_ip):57        path = 'domains/{}/records'.format(self._domain_name)58        json_data = {'type': 'A', 'name': host_name, 'data': actual_ip}59        response = self._api.post(path, json_data)60        if not response.ok:61            raise DomainRecordError('could not create domain record')62    def find(self, host_name):63        path = 'domains/{}/records?per_page=100'.format(self._domain_name)64        response = self._api.get(path)65        while True:66            if not response.ok:67                raise DomainRecordError('failure when finding domain record')68            result = response.json()69            for domain_record in result['domain_records']:70                if (domain_record['type'] == 'A' and71                        domain_record['name'] == host_name):72                    return {'id': domain_record['id'],73                            'ip': domain_record['data']}74            if result['links'] and 'next' in result['links']:75                response = self._api.follow_link(result['links']['next'])76            else:77                return None78    def update(self, domain_record, actual_ip):79        path = 'domains/{}/records/{}'.format(self._domain_name,80                                              domain_record['id'])81        json_data = {'data': actual_ip}82        response = self._api.put(path, json_data)83        if not response.ok:...

main_cms_service_entity.py

Source:main_cms_service_entity.py

1from common.global_config import Global2from service_entities.cms_entities.app_cache_update_entity import AppCacheUpdateEntity3from service_entities.cms_entities.pdcmanager_doinsertproductbasecfg_entity import \4    PdcmanagerDoinsertproductbasecfgEntity5from service_entities.cms_entities.pdcmanager_doproductbasechecklist_entity import \6    PdcmanagerDoproductbasechecklistEntity7from service_entities.cms_entities.pdcmanager_doproductbasesubcheck_entity import \8    PdcmanagerDoproductbasesubcheckEntity9from service_entities.cms_entities.pdcmanager_saveproductdetailcof_entity import \10    PdcmanagerSaveproductdetailcofEntity11from service_entities.cms_entities.pdcmanager_updateproductchannle_entity import \12    PdcmanagerUpdateproductchannleEntity13from service_entities.cms_entities.prodquota_add_entity import ProdquotaAddEntity14from service_entities.cms_entities.prodquota_changecount_entity import ProdquotaChangecountEntity15from service_entities.cms_entities.prodquota_changequato_entity import ProdquotaChangequatoEntity16from service_entities.cms_entities.user_login_entity import UserLoginEntity17ENVIRONMENT_MAP_DOMAIN_NAME = {18    "uat": Global.Environment.HUXIN_CMS_UAT,19}20class MainCmsServiceEntity():21    def __init__(self, environment='uat'):22        self._domain_name = ENVIRONMENT_MAP_DOMAIN_NAME[environment]23    @property24    def current_product_id(self):25        return self._current_product_id26    def login(self, **kwargs):27        entity = UserLoginEntity(self._domain_name)28        entity.send_request(**kwargs)29        return entity30    def new_product(self, **kwargs):31        entity = PdcmanagerDoinsertproductbasecfgEntity(self._domain_name)32        entity.send_request(**kwargs)33        self._current_product_id = entity.productid34        return entity35    def submit_product(self, **kwargs):36        entity = PdcmanagerDoproductbasesubcheckEntity(self._domain_name)37        entity.send_request(**kwargs)38        return entity39    def check_product(self, **kwargs):40        entity = PdcmanagerDoproductbasechecklistEntity(self._domain_name)41        entity.send_request(**kwargs)42        return entity43    def update_product_channle(self, **kwargs):44        entity = PdcmanagerUpdateproductchannleEntity(self._domain_name)45        entity.send_request(**kwargs)46        return entity47    def save_product_detail(self, **kwargs):48        entity = PdcmanagerSaveproductdetailcofEntity(self._domain_name)49        entity.send_request(**kwargs)50        return entity51    def add_product_quota(self, **kwargs):52        entity = ProdquotaAddEntity(self._domain_name)53        entity.send_request(**kwargs)54        return entity55    def change_product_quota(self, **kwargs):56        entity = ProdquotaChangequatoEntity(self._domain_name)57        entity.send_request(**kwargs)58        return entity59    def change_product_count(self, **kwargs):60        entity = ProdquotaChangecountEntity(self._domain_name)61        entity.send_request(**kwargs)62        return entity63    def app_cache_update(self, **kwargs):64        entity = AppCacheUpdateEntity(self._domain_name)65        entity.send_request(**kwargs)...

domain_expiration.py

Source:domain_expiration.py

1import requests2import whois3import re4import subprocess5from bs4 import BeautifulSoup6from datetime import datetime, timedelta7dt = datetime.now()8print("Today's date is " + str(dt))9def query_domain(_domain_name):10    domain = whois.Domain11    try:12        domain = whois.query(_domain_name)13    except Exception as e:14        # Try running whois in commandline and return a dict15        print(f"Using whois command for {_domain_name}")16        output = subprocess.run(["whois", _domain_name], capture_output=True)17        result = re.findall(r'[\n\r].*Registry Expiry Date:\s*([^\n\r]*)', output.stdout.decode())18        for r in result:19            format = "%Y-%m-%dt%H:%M:%S.%fz"20            time = datetime.strptime(r, format).astimezone().replace(tzinfo=None)21            domain.expiration_date = time22    return domain23def check_domain_expiry(_domain_name, package_name):24    try:25        domain = query_domain(_domain_name)26        if domain is None or domain.expiration_date is None:27            print(f"[â] {package_name} || {_domain_name} (Library is unable to retrieve expiration date info)")28        elif dt > domain.expiration_date:29            print(f"[ð¨] {package_name} || {_domain_name} has expired since " + str(domain.expiration_date))30        else:31            print(f"[â] {package_name} || {_domain_name} is valid till " + str(domain.expiration_date))32    except Exception as e:33        print(e.__str__(), _domain_name)34def sanitize_non_domain_chars(domain_name):35    return domain_name.replace(">", "").rstrip()36def extract_email_domains(package_url):37    email_domains = []38    response = requests.get(package_url)39    soup = BeautifulSoup(response.text, 'html.parser')40    mailtos = list(dict.fromkeys(soup.select('a[href^=mailto]')))41    for mail_list in mailtos:42        href = mail_list['href'].replace("%40", "@")43        if href.__contains__(","):44            # Contains more than one email45            emails = href.split(":")[1].split(",")46            for email in emails:47                email_domains.append(sanitize_non_domain_chars(email.split('@')[1]))48        else:49            try:50                prefix, domain = href.split('@')51            except ValueError:52                print("Error: " + href)53                break54            email_domains.append(sanitize_non_domain_chars(domain))55    return email_domains56with open('requirements.txt', 'r') as packages:57    lines = packages.readlines()58    for line in lines:59        if line.__contains__("=="):60            package_name = line.rstrip().split('==')[0]61            for domain_name in extract_email_domains(f"https://pypi.org/project/{package_name}"):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.