Compare commits
32 Commits
anticaptch
...
master
|
@ -1,4 +1,5 @@
|
|||
.idea
|
||||
.vscode
|
||||
**/__pycache__/
|
||||
**/main.py
|
||||
**/main.py
|
||||
*.http
|
|
@ -12,7 +12,7 @@ class AntiCaptcha:
|
|||
ep: str = ""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_json_post__ (self, path: str, object: dict):
|
||||
"""
|
||||
|
|
|
@ -0,0 +1,106 @@
|
|||
import base64
|
||||
import json
|
||||
import os
|
||||
import requests
|
||||
|
||||
class API_Proxy:
|
||||
"""
|
||||
## API Proxy
|
||||
---
|
||||
O Serviço API Proxy permite realizar chamadas a api externas proxiadas pelo Replay. Dessa forma, as chamadas podem ficar pré- configuradas dentro do Replay.
|
||||
|
||||
Caso o robô ou aplicação desejem alterar algum parâmetro antes da chamada ainda é possível, para isso, deve-se modificar a estrutura da requisição.
|
||||
|
||||
Para entender melhor a como funciona a estrutura de requisição, você pode visitar a documentaçao oficial ou dar uma olhada no método "do ()" deste client.
|
||||
"""
|
||||
|
||||
ep: str = ""
|
||||
|
||||
APIProxyRequest = {
|
||||
"Name": None,
|
||||
"Method": None,
|
||||
"Url": None,
|
||||
"Header": None,
|
||||
"Body": None,
|
||||
"Readonly": None
|
||||
}
|
||||
|
||||
APIProxyResponse = {
|
||||
"Status": None,
|
||||
"StatusCode": None,
|
||||
"Header": None,
|
||||
"Body": None
|
||||
}
|
||||
|
||||
def __init__ (self):
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_json_post__(self, path: str, object: dict):
|
||||
|
||||
"""
|
||||
## HTTP JSON POST
|
||||
---
|
||||
Este método é responsável por realizar requisições HTTP do tipo POST para objetos JSON.
|
||||
|
||||
Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
|
||||
"""
|
||||
|
||||
url = self.ep + path
|
||||
print("Calling: " + url)
|
||||
|
||||
apikey = os.environ.get('REPLAY_APIKEY')
|
||||
headers = {"X-API-KEY": apikey}
|
||||
res = requests.post(url, json = object, headers = headers, verify = False)
|
||||
|
||||
if res.status_code >= 400:
|
||||
raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
|
||||
|
||||
|
||||
return json.loads(res.text)
|
||||
|
||||
|
||||
def do (self, request: dict):
|
||||
"""
|
||||
## API Proxy Do
|
||||
Faz o proxy da chamada remota e retorna qualquer que seja o resultado desta requisição.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- request: Estrutura APIProxyRequest definida nesta mesma classe. Instancie-a e altere os valores das chaves de acordo com sua necessidade, veja abaixo como:
|
||||
|
||||
{
|
||||
"Name": "Nome de template da chamada",
|
||||
|
||||
"Method": "Método de requisição (GET, POST, PUT, ...)",
|
||||
|
||||
"Url": "Url da API que se deseja chamar",
|
||||
|
||||
"Header": {
|
||||
"Chave da configuração de cabeçalho": "Valor associado",
|
||||
|
||||
...
|
||||
},
|
||||
|
||||
"Body": "Elementos pertencentes ao corpo da requisição",
|
||||
|
||||
"Readonly":
|
||||
|
||||
- True: Mesmo que seja encontrado um template correspondente ao valor de "Name" passado, as configurações desta estrutura prevalecerão na requisição.
|
||||
|
||||
- False: Caso um template seja encontrado em correspondência ao valor passado em "Name", as configurações do mesmo sobrescreverão as informações passadas nesta estrutura.
|
||||
}
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> Estrutura APIProxyResponse já definida nesta classe.
|
||||
"""
|
||||
|
||||
do_response = self.__request_json_post__ ("/ipc/apiproxy/do", request)
|
||||
|
||||
do_response["body"] = base64.b64decode(do_response["body"])
|
||||
do_response["body"] = str(do_response["body"], "utf-8")
|
||||
|
||||
if do_response["header"]["Content-Type"].find("json") != -1:
|
||||
do_response["body"] = json.loads(do_response["body"])
|
||||
|
||||
return do_response
|
|
@ -0,0 +1,14 @@
|
|||
Feature: The API Proxy client
|
||||
Scenario: A HTTP request with json body response
|
||||
Given a API Proxy client
|
||||
And an APIProxyRequest instance filled with the method GET for the API url: https://pokeapi.co/api/v2/pokemon/ditto
|
||||
When a request is made with the previous instance
|
||||
Then the body of this response must be a python dictionary
|
||||
And the pokemon name, contained in the response body, must be ditto
|
||||
|
||||
|
||||
Scenario: A HTTP request with text body response
|
||||
Given a API Proxy client
|
||||
And an APIProxyRequest instance filled with the method GET for the API url: https://www.slashdot.org
|
||||
When a request is made with the previous instance
|
||||
Then the response body content type must be string
|
|
@ -0,0 +1,39 @@
|
|||
from re import A
|
||||
from behave import *
|
||||
from cli import API_Proxy
|
||||
|
||||
# =========================== JSON RESPONSE BODY ===========================
|
||||
@given(u'a API Proxy client')
|
||||
def step_impl(context):
|
||||
context.client = API_Proxy ()
|
||||
|
||||
@given(u'an APIProxyRequest instance filled with the method GET for the API url: https://pokeapi.co/api/v2/pokemon/ditto')
|
||||
def step_impl(context):
|
||||
context.request = context.client.APIProxyRequest
|
||||
context.request["Method"] = "GET"
|
||||
context.request["Url"] = "https://pokeapi.co/api/v2/pokemon/ditto"
|
||||
|
||||
@when(u'a request is made with the previous instance')
|
||||
def step_impl(context):
|
||||
context.response = context.client.do (context.request)
|
||||
|
||||
@then(u'the body of this response must be a python dictionary')
|
||||
def step_impl(context):
|
||||
assert type (context.response["body"]) == dict, "Something went wrong calling the pokemon/ditto API."
|
||||
|
||||
@then(u'the pokemon name, contained in the response body, must be ditto')
|
||||
def step_impl(context):
|
||||
assert context.response["body"]["forms"][0]["name"] == "ditto", "This pokemon is not Ditto."
|
||||
|
||||
|
||||
|
||||
# =========================== TEXT RESPONSE BODY ===========================
|
||||
@given(u'an APIProxyRequest instance filled with the method GET for the API url: https://www.slashdot.org')
|
||||
def step_impl(context):
|
||||
context.request = context.client.APIProxyRequest
|
||||
context.request["Method"] = "GET"
|
||||
context.request["Url"] = "https://www.slashdot.org"
|
||||
|
||||
@then(u'the response body content type must be string')
|
||||
def step_impl(context):
|
||||
assert type (context.response["body"]) == str, "Something went wrong calling the slashdot API."
|
|
@ -16,7 +16,7 @@ class AutoHotKey:
|
|||
ep: str = ""
|
||||
|
||||
def __init__ (self):
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_raw_post__ (self, path: str, data: str = ""):
|
||||
|
||||
|
|
|
@ -16,7 +16,7 @@ class AutoIt:
|
|||
ep: str = ""
|
||||
|
||||
def __init__ (self):
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_raw_post__ (self, path: str, data: str = ""):
|
||||
|
||||
|
|
|
@ -21,7 +21,7 @@ class Chrome:
|
|||
ep: str = ""
|
||||
|
||||
def __init__(self):
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_get__(self, data: str):
|
||||
|
||||
|
|
|
@ -0,0 +1,148 @@
|
|||
from datetime import datetime
|
||||
import json
|
||||
|
||||
import requests
|
||||
|
||||
class DataAPI:
|
||||
"""
|
||||
## Data API
|
||||
---
|
||||
Esta classe permite a execução de comandos básicos SQL para o banco de dados remoto do Replay.
|
||||
|
||||
Para que a classe funcione corretamente, é necessário que o usuário tenha acesso à chave da API do BD remoto. Devendo passá-la como parâmetro ao instanciar a classe DataAPI. Por exemplo:
|
||||
d_api = DataAPI ("ahsgcn21390-dsaf_dain2eq81288sad9")
|
||||
"""
|
||||
|
||||
ep: str = ""
|
||||
|
||||
def __init__ (self, api_key: str):
|
||||
self.ep = "https://dataapi.digitalcircle.com.br"
|
||||
self.api_key = api_key
|
||||
|
||||
def __request_json_post__ (self, object: dict):
|
||||
|
||||
"""
|
||||
## HTTP JSON POST
|
||||
---
|
||||
Este método é responsável por realizar requisições HTTP do tipo POST para objetos JSON.
|
||||
|
||||
Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
|
||||
"""
|
||||
|
||||
url = self.ep + "/"
|
||||
print("Calling: " + url)
|
||||
|
||||
headers = {"X-API-KEY": self.api_key}
|
||||
res = requests.post(url, json = object, headers = headers, verify = False)
|
||||
|
||||
if res.status_code >= 400:
|
||||
raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
|
||||
if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
|
||||
return json.loads(res.text)
|
||||
else:
|
||||
return res.text
|
||||
|
||||
def do (self, dataAPI_request: dict):
|
||||
"""
|
||||
## DataAPI Do
|
||||
Este método é responsável pela execução dos comandos SQL no Banco de Dados remoto.
|
||||
|
||||
---
|
||||
#### Parâmetros de dataAPI_request:
|
||||
- Col: Nome da tabela SQL em que se quer realizar a operação (OBRIGATÓRIO)
|
||||
- Op: Operação que se quer desenvolver. São elas: "R", "C", "D", "U" (OBRIGATÓRIO)
|
||||
- R: Retrieve (Busca informações da coluna da tabela)
|
||||
- C: Create (Cria uma nova tabela)
|
||||
- D: Delete (Deleta uma tabela ou as colunas de uma tabela)
|
||||
- U: Update (Atualiza as informações de uma tabela)
|
||||
|
||||
- Q: Nome do atributo (coluna) que se quer consultar ou alterar na tabela SQL
|
||||
- Id: Identificador da instância (linha) da tabela que se quer realizar a operação
|
||||
- Data: Dados que se quer inserir na tabela
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> O retorno do comando feito, qualquer que seja ele.
|
||||
"""
|
||||
|
||||
return self.__request_json_post__(dataAPI_request)
|
||||
|
||||
def registrar_exec (self, register_table: str, check: bool):
|
||||
"""
|
||||
## Registrar Execução
|
||||
Esta função é utilizada para realizar um registro de execuções falhas e bem sucedidas de um robô. Tudo é feito diretamente na Base de Dados remota, eliminando a necessidade de planilhas Excel locais.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- register_table: Tabela utilizada para os registros de execução. Para começos de projeto, procure criar uma nova tabela, cheque se ela já existe utilizando a operação de Retrieve pelo método Do.
|
||||
- check: Assume dois valores dependendo do sucesso da execução do robô: True ou False.
|
||||
- True: Execução obteve sucesso.
|
||||
- False: Execução falhou em algum ponto.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
---
|
||||
"""
|
||||
|
||||
today_date = datetime.now().isoformat()[:10]
|
||||
|
||||
table_check = self.do({
|
||||
"Col": register_table,
|
||||
"Op": "R",
|
||||
"Q": f"@[?date=='{today_date}']"
|
||||
})
|
||||
|
||||
if len(table_check["data"]) > 0:
|
||||
data = table_check["data"][0]
|
||||
id = str(int(data["ID"]))
|
||||
|
||||
if check:
|
||||
success = int(data["exec"]) + 1
|
||||
|
||||
try:
|
||||
self.do({
|
||||
"Col": register_table,
|
||||
"Op": "U",
|
||||
"Id": id,
|
||||
"Data": {
|
||||
"exec": success
|
||||
}
|
||||
})
|
||||
except:
|
||||
return "Algo falhou ao registrar a execução"
|
||||
|
||||
else:
|
||||
fail = int(data["err"]) + 1
|
||||
|
||||
try:
|
||||
self.do({
|
||||
"Col": register_table,
|
||||
"Op": "U",
|
||||
"Id": id,
|
||||
"Data": {
|
||||
"err": fail
|
||||
}
|
||||
})
|
||||
except:
|
||||
return "Algo falhou ao registrar a execução"
|
||||
|
||||
else:
|
||||
if check:
|
||||
success = 1
|
||||
fail = 0
|
||||
else:
|
||||
success = 0
|
||||
fail = 1
|
||||
|
||||
try:
|
||||
self.do({
|
||||
"Col": register_table,
|
||||
"Op": "C",
|
||||
"Data": {
|
||||
"date": today_date,
|
||||
"exec": success,
|
||||
"err": fail,
|
||||
}
|
||||
})
|
||||
except:
|
||||
return "Algo falhou ao registrar a execução"
|
|
@ -0,0 +1,13 @@
|
|||
Feature: The Data API Client
|
||||
Scenario: Testing the data_API.do() method
|
||||
Given a data_API client
|
||||
When a new table is created with the value "Bond" for the query "the_name_is"
|
||||
Then the value retrieved from the new table associated with the query "the_name_is" must be "Bond"
|
||||
|
||||
|
||||
Scenario: Testing the data_API.registrar_exec method
|
||||
Given a data_API client
|
||||
When the registrar_exec method is called with True for check
|
||||
And the registrar_exec method is called with False for check
|
||||
And data is retrieved from the table
|
||||
Then there must be at least 1 register for "exec" and "err" today
|
|
@ -0,0 +1,53 @@
|
|||
from behave import *
|
||||
from cli import *
|
||||
|
||||
# ============================= DATA_API DO =============================
|
||||
@given(u'a data_API client')
|
||||
def step_impl(context):
|
||||
context.client = DataAPI ("Insira aqui a chave de API. Você pode obtê-la pedindo ao fornecedor deste client.")
|
||||
|
||||
@when(u'a new table is created with the value "Bond" for the query "the_name_is"')
|
||||
def step_impl(context):
|
||||
context.client.do ({
|
||||
"Col": "newTestTable",
|
||||
"Op": "C",
|
||||
"Data": {
|
||||
"the_name_is": "Bond",
|
||||
"James": "Bond"
|
||||
}
|
||||
})
|
||||
|
||||
@then(u'the value retrieved from the new table associated with the query "the_name_is" must be "Bond"')
|
||||
def step_impl(context):
|
||||
context.retrieved_data = context.client.do({
|
||||
"Col": "newTestTable",
|
||||
"Op": "R"
|
||||
})["data"][0]
|
||||
|
||||
context.query_value = context.retrieved_data["the_name_is"]
|
||||
|
||||
assert context.query_value == "Bond", "Something went wrong with the method Do"
|
||||
|
||||
|
||||
# ============================= DATA_API REGISTRAR_EXEC =============================
|
||||
@when(u'the registrar_exec method is called with True for check')
|
||||
def step_impl(context):
|
||||
context.client.registrar_exec("anotherNewTestTable", True)
|
||||
|
||||
@when(u'the registrar_exec method is called with False for check')
|
||||
def step_impl(context):
|
||||
context.client.registrar_exec("anotherNewTestTable", False)
|
||||
|
||||
@when(u'data is retrieved from the table')
|
||||
def step_impl(context):
|
||||
today_date = datetime.now().isoformat()[:10]
|
||||
|
||||
context.retrieved_data = context.client.do({
|
||||
"Col": "anotherNewTestTable",
|
||||
"Op": "R",
|
||||
"Q": f"@[?date=='{today_date}']"
|
||||
})["data"][0]
|
||||
|
||||
@then(u'there must be at least 1 register for "exec" and "err" today')
|
||||
def step_impl(context):
|
||||
assert context.retrieved_data["exec"] > 0 and context.retrieved_data["err"] > 0, "Something went wrong with the method registrar_exec"
|
|
@ -13,7 +13,7 @@ class Excel:
|
|||
ep: str = ""
|
||||
|
||||
def __init__ (self):
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_json_post__ (self, path: str, object: dict):
|
||||
|
||||
|
|
|
@ -13,7 +13,7 @@ class Ocr:
|
|||
ep: str = ""
|
||||
|
||||
def __init__ (self):
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_json_post__ (self, path: str, object: dict):
|
||||
|
||||
|
|
|
@ -0,0 +1,547 @@
|
|||
from datetime import datetime
|
||||
import json
|
||||
import os
|
||||
import requests
|
||||
import urllib
|
||||
|
||||
class Replay:
|
||||
"""
|
||||
## Replay
|
||||
---
|
||||
Esta classe é utilizada para fazer manipulações com o Replay instalado na máquina. Ela concede acesso às diferentes funcionalidades que o mesmo apresenta na interface do Console, sem que o programador precise realizar WebScraping para acessá-lo.
|
||||
|
||||
Permite por exemplo a extração de configurações definidas para certos repositórios, registro de Logs, manipulação de banco de dados (SQLite), entre várias outras coisas.
|
||||
"""
|
||||
|
||||
ep: str = ""
|
||||
|
||||
def __init__ (self):
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __request_get__ (self, data: str):
|
||||
|
||||
"""
|
||||
## HTTP GET
|
||||
---
|
||||
Este método é responsável por realizar requisições HTTP do tipo GET.
|
||||
|
||||
Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
|
||||
"""
|
||||
|
||||
url = self.ep + data
|
||||
print("Calling: " + url)
|
||||
|
||||
apikey = os.environ.get('REPLAY_APIKEY')
|
||||
headers = {"X-API-KEY": apikey}
|
||||
res = requests.get(url, headers=headers, verify=False)
|
||||
|
||||
if res.status_code >= 400:
|
||||
raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
|
||||
if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
|
||||
return json.loads(res.text)
|
||||
else:
|
||||
return res.text
|
||||
|
||||
def __request_json_post__ (self, path: str, object: dict):
|
||||
|
||||
"""
|
||||
## HTTP JSON POST
|
||||
---
|
||||
Este método é responsável por realizar requisições HTTP do tipo POST para objetos JSON.
|
||||
|
||||
Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
|
||||
"""
|
||||
|
||||
url = self.ep + path
|
||||
print("Calling: " + url)
|
||||
|
||||
apikey = os.environ.get('REPLAY_APIKEY')
|
||||
headers = {"X-API-KEY": apikey}
|
||||
res = requests.post(url, json = object, headers = headers, verify = False)
|
||||
|
||||
if res.status_code >= 400:
|
||||
raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
|
||||
if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
|
||||
return json.loads(res.text)
|
||||
else:
|
||||
return res.text
|
||||
|
||||
def __request_raw_post__ (self, path: str, data: str):
|
||||
|
||||
"""
|
||||
## HTTP RAW POST
|
||||
---
|
||||
Este método é responsável por realizar requisições HTTP do tipo POST para objetos RAW.
|
||||
|
||||
Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
|
||||
"""
|
||||
|
||||
url = self.ep + path
|
||||
print("Calling: " + url)
|
||||
|
||||
apikey = os.environ.get('REPLAY_APIKEY')
|
||||
headers = {"X-API-KEY": apikey}
|
||||
res = requests.post(url, data = data, headers = headers, verify = False)
|
||||
|
||||
if res.status_code >= 400:
|
||||
raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
|
||||
if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
|
||||
return json.loads(res.text)
|
||||
else:
|
||||
return res.text
|
||||
|
||||
def replay_env_queue_id (self):
|
||||
"""
|
||||
## Replay Environment Queue ID
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_QUEUEID", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_QUEUEID")
|
||||
|
||||
def replay_env_alias (self):
|
||||
"""
|
||||
## Replay Environment Alias
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ALIAS", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_ALIAS")
|
||||
|
||||
def replay_env_feature_dir (self):
|
||||
"""
|
||||
## Replay Environment Feature Directory
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_FEATUREDIR", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_FEATUREDIR")
|
||||
|
||||
def replay_env_root (self):
|
||||
"""
|
||||
## Replay Environment Root
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ROOT", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_ROOT")
|
||||
|
||||
def replay_env_addr (self):
|
||||
"""
|
||||
## Replay Environment Address
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ADDR", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_ADDR")
|
||||
|
||||
def replay_env_home_dir (self):
|
||||
"""
|
||||
## Replay Environment Home Directory
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_HOMEDIR", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_HOMEDIR")
|
||||
|
||||
def replay_env_repo (self):
|
||||
"""
|
||||
## Replay Environment Repository
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_REPO", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_REPO")
|
||||
|
||||
def replay_env_ver (self):
|
||||
"""
|
||||
## Replay Environment Version
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_VER", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_VER")
|
||||
|
||||
def replay_env_repo_dir (self):
|
||||
"""
|
||||
## Replay Environment Repository Directory
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_REPODIR", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_REPODIR")
|
||||
|
||||
def replay_env_data_dir (self):
|
||||
"""
|
||||
## Replay Environment Data Directory
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_DATADIR", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_DATADIR")
|
||||
|
||||
def replay_env_instance_alias (self):
|
||||
"""
|
||||
## Replay Environment Instance Alias
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_INSTANCE_ALIAS", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_INSTANCE_ALIAS")
|
||||
|
||||
def replay_env_api_key (self):
|
||||
"""
|
||||
## Replay Environment API Key
|
||||
---
|
||||
Esta função retorna o valor atribuído à variável de ambiente "REPLAY_APIKEY", caso exista.
|
||||
"""
|
||||
|
||||
return os.environ.get ("REPLAY_APIKEY")
|
||||
|
||||
def sql (self, command:str):
|
||||
"""
|
||||
## SQL Commands Executer
|
||||
Este método faz um requisição SQL no Replay na aba "Banco de Dados"
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- command: Comando SQL que se deseja executar.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> Qualquer que seja o retorno do comando executado.
|
||||
"""
|
||||
|
||||
sql = {
|
||||
"Sql": command
|
||||
}
|
||||
|
||||
return self.__request_json_post__("/api/v1/sql", sql)
|
||||
|
||||
def new_log (self):
|
||||
"""
|
||||
## New Log
|
||||
Este método faz a instanciação de um objeto do tipo "Log" para que ele possa ser registrado no Replay.
|
||||
"""
|
||||
log = {
|
||||
"Alias": str (self.replay_env_alias()),
|
||||
"Dtlog": datetime.now().isoformat() + "Z"
|
||||
}
|
||||
|
||||
return log
|
||||
|
||||
def log (self, log: dict):
|
||||
"""
|
||||
## Log
|
||||
Esta função realiza a postagem de um Log na base de dados do replay, podendo ser visualizado pelo Console.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- log: Este parâmetro é uma instância da classe Logs que deve ser criada a partir do método "new_logs".
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
---
|
||||
"""
|
||||
log["Dtlogend"] = datetime.now().isoformat() + "Z"
|
||||
|
||||
return self.__request_json_post__ ("/api/v1/log/add", log)
|
||||
|
||||
def config_get (self, config_name: str):
|
||||
"""
|
||||
## Configuration Getter
|
||||
Este método busca a configuração associada ao nome recebido no repositório atual.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- config_name: Nome da configuração que se quer procurar o valor.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> O valor associado à configuração procurada.
|
||||
"""
|
||||
|
||||
config_name = urllib.parse.quote(config_name, "")
|
||||
|
||||
return self.__request_get__(f"/api/v1/config/byrepo/{int(os.path.basename(os.getcwd()))}/{config_name}")
|
||||
|
||||
def config_get_global (self, config_name: str):
|
||||
"""
|
||||
## Global Configuration Getter
|
||||
Este método procura entre as configurações globais, uma correspondência ao nome de configuração recebido.
|
||||
|
||||
---
|
||||
### Obs:
|
||||
Uma configuração global é definida diretamente na aba de Configurações do Console do Replay.
|
||||
|
||||
Deve-se clicar em "Adicionar" e digitar o nome da configuração, depois editar o nome da mesma.
|
||||
|
||||
Se uma configuração não está associada a nenhum repositório em específico, isso quer dizer que ela é global.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- config_name: Nome da configuração global que se quer saber o valor.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> O valor da configuração encontrada.
|
||||
-> "" caso a configuração não exista.
|
||||
"""
|
||||
|
||||
config_name = urllib.parse.quote(config_name, "")
|
||||
|
||||
return self.__request_get__("/api/v1/config/global/"+config_name)
|
||||
|
||||
def config_get_all (self):
|
||||
"""
|
||||
## All Configuration Getter
|
||||
Este método busca por todas as configurações contidas no repositório atual.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> Uma lista de dicionários contendo todas as configs e informações sobre elas. Por exemplo:
|
||||
[
|
||||
{
|
||||
"id": número associado a config na base de dados do replay,
|
||||
|
||||
"k": nome associado à config na base de dados do replay,
|
||||
|
||||
"v" valor atribuído à config,
|
||||
|
||||
"t": tipo da config em questão (que pode ser 't', 'p' ou 'json'),
|
||||
|
||||
"repo": repositório ao qual a configuração está associada
|
||||
}
|
||||
]
|
||||
"""
|
||||
|
||||
return self.__request_get__(f"/api/v1/config/byrepo/{int(os.path.basename(os.getcwd()))}")
|
||||
|
||||
def config_get_map (self) -> dict[str, dict]:
|
||||
"""
|
||||
## Configuration Getter Mapper
|
||||
Este método executa o método "config_get_all" para conseguir todas as configurações alocadas para o reposiório do projeto atual. Após isso, cria um dicionário de strings para outros dicionários contendo as configurações recebidas. As strings do primeiro dicionário são o nome associado à configuração.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> Dicionário de strings (nome da configuração) para dicionários (todas as propriedades da configuração).
|
||||
"""
|
||||
|
||||
dict_configs = {}
|
||||
configs = self.config_get_all()
|
||||
|
||||
for config in configs:
|
||||
dict_configs[config["k"]] = config
|
||||
|
||||
return dict_configs
|
||||
|
||||
def open_app (self, id: int):
|
||||
"""
|
||||
## Open Apps
|
||||
Este método realiza a execução de um aplicativo cujo ID foi recebido.
|
||||
|
||||
---
|
||||
### Obs:
|
||||
É importante ressaltar que a execução do aplicativo lançado é síncrona, ou seja, o código não terá sua execução completada antes da finalização do aplicativo.
|
||||
|
||||
Para alguns casos de uso, a execução síncrona é bastante útil, mas para outros, pode dificultar bastante o processo de desenvolvimento.
|
||||
|
||||
Sendo assim, é possível utilizar o recurso de multiThreading do python para que o aplicativo seja executado assíncronamente. Isso é feito através da biblioteca "threading", que não é nativa. Deve-se instalá-la e utilizar de acordo com a necessidade.
|
||||
|
||||
Para maiores informações sobre a biblioteca threading, visite o site:
|
||||
http://pymotw.com/2/threading/
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- id: Identificador do aplicativo que se deseja executar.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
---
|
||||
"""
|
||||
|
||||
return self.__request_get__(f"/api/v1/app/run/{id}")
|
||||
|
||||
def cron_reload (self):
|
||||
"""
|
||||
## Cronometer Reload
|
||||
Este método está ligado à página de Agendamentos no Console do Replay. Ele recarrega esta funcionalidade para atualizar os registros de agendamento.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> ???
|
||||
"""
|
||||
|
||||
return self.__request_get__("/ipc/cron/reload")
|
||||
|
||||
def menu_get_all_enabled (self):
|
||||
"""
|
||||
Get All Enabled Menus
|
||||
Este método está relacionado à aba de "Menus" do Console do Replay. Todos os menus cadastrados naquela aba e que tenham a checkbox do lado esquerdo marcada deverão ser buscados por este método.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> Lista de dicionários, onde os dicionários são os dados/informações dos menus encontrados.
|
||||
"""
|
||||
|
||||
return self.__request_get__("/api/v1/menu/enabled")
|
||||
|
||||
def service_stop_all (self):
|
||||
"""
|
||||
## Stop All Services
|
||||
Este método realiza a parada forçada de todos os serviços que estiverem ativos. É possível visualizar quais serviços estão ativos pela aba "Serviços" do Console do Replay. Todos cujo PID tenha um valor diferente de -1 e 0 são serviços que estão, atualmente, ativos.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> "ok" caso o método tenha sido executado corretamente
|
||||
"""
|
||||
|
||||
return self.__request_get__("/api/v1/service/op/stopall")
|
||||
|
||||
def exit (self):
|
||||
"""
|
||||
## Exit
|
||||
Finaliza a sessão do Replay. Equivalente a clicar no ícone e depois em "Quit".
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> "ok" caso o método tenha sido executado corretamente.
|
||||
"""
|
||||
|
||||
try:
|
||||
self.__request_get__("/api/v1/exit")
|
||||
except:
|
||||
return "ok"
|
||||
|
||||
def queue_add (self, job_id: int, params: dict = {}):
|
||||
"""
|
||||
## Queue Add
|
||||
Este método adiciona um novo "job" à fila de execução. É possível ver o resultado na aba "Fila" presente na interface do Console do Replay.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- job_id: Identificador do robô, aplicativo ou serviço que se deve adicionar à fila de execução.
|
||||
- params: Parâmetros de execução do robô. Ao rodar um robô pela interface do Replay, é aberto um editor de Objetos JSON. Esse parâmetro é basicamente o editor. Deve ser passado um dicionário contendo todos os parâmetros que se quer informar para a execução do robô.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> "ok" caso a adição à fila tenha ocorrido com sucesso.
|
||||
"""
|
||||
|
||||
object = json.dumps(params)
|
||||
|
||||
return self.__request_raw_post__(f"/api/v1/queue/add/{job_id}", object)
|
||||
|
||||
def queue_get_data (self, queue_id: int):
|
||||
"""
|
||||
## Queue Get Data
|
||||
Este método busca informações sobre um "job" presente na fila.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- queue_id: Identificador do robô na fila que se quer obter os dados. Para obter esta informação, você pode utilizar o método "replay_env_queue_id()".
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> String emulando um objeto JSON com todos os dados passados ao robô no momento da execução (Editor de Objetos JSON).
|
||||
"""
|
||||
|
||||
return self.__request_get__(f"/api/v1/queue/getrequest/{queue_id}")
|
||||
|
||||
def queue_get_my_data (self):
|
||||
"""
|
||||
## Queue Get My Data
|
||||
Este método é uma extensão do método "queue_get_data". Ele elimina a necessidade de se adicionar o identificador do robô na fila, já que a variável de ambiente é buscada dentro do método. Além disso, é interessante ressaltar que este método faz a busca de informações sobre o robô que está atualmente rodando.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> String emulando um objeto JSON com todos os dados passados ao robô no momento da execução (Editor de Objetos JSON).
|
||||
"""
|
||||
|
||||
my_data = self.replay_env_queue_id()
|
||||
return self.queue_get_data (my_data)
|
||||
|
||||
def queue_get_my_json (self):
|
||||
"""
|
||||
## Queue Get My JSON
|
||||
Este método é uma extensão do método "queue_get_my_data", mas ele faz com que o objeto seja um dicionário ao invés de uma string. Além disso, é interessante ressaltar que este método faz a busca de informações sobre o robô que está atualmente rodando.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
#### Retorna:
|
||||
-> Dicionário emulando um objeto JSON com todos os dados passados ao robô no momento da execução (Editor de Objetos JSON).
|
||||
"""
|
||||
|
||||
str_obj = self.queue_get_my_data()
|
||||
|
||||
return json.loads(str_obj)
|
||||
|
||||
def queue_abort (self):
|
||||
"""
|
||||
## Queue Abort
|
||||
Este método é utilizado para abortar o processo atual que está sendo executado na fila. É equivalente ao botão "Abortar" presente na aba de "Fila" no Console do Replay.
|
||||
|
||||
É possível abortar mais de um executável na fila. Basta adicioná-lo, esperar que seja executado e então utilizar o método abort.
|
||||
|
||||
#### Lembre-se sempre que este método somente surte efeito no processo ATUAL executado pela fila.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
---
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
- "ok": caso tudo tenha ocorrido corretamente e o método tenha abortado um elemento da fila.
|
||||
- None: caso o método tenha sido executado e não haja mais nenhum robô na fila sendo executado atualmente.
|
||||
"""
|
||||
|
||||
try:
|
||||
self.__request_get__("/api/v1/queue/abort")
|
||||
except:
|
||||
return "ok"
|
||||
|
||||
def queue_enqueue (self, job_id: int):
|
||||
"""
|
||||
## Queue Enqueue
|
||||
Este método adiciona a fila um robô cujo ID tenha sido passado como parâmetro.
|
||||
|
||||
---
|
||||
#### Parâmetros:
|
||||
- job_id: Identificador do robô que se quer adicionar à fila na base de dados do replay.
|
||||
|
||||
---
|
||||
#### Retorna:
|
||||
-> "ok" caso o robô tenha sido adicionado à fila corretamente.
|
||||
"""
|
||||
|
||||
return self.__request_get__(f"/api/v1/robots/op/enqueue/{job_id}")
|
|
@ -0,0 +1,41 @@
|
|||
Feature: The Replay API client
|
||||
|
||||
Scenario: Testing every environment variable getter
|
||||
Given the Replay client
|
||||
When each one of the getters is called
|
||||
Then all of the returns must be either of the type string or the type None
|
||||
|
||||
|
||||
Scenario: Testing the process of creating and registering a Log
|
||||
Given the Replay client
|
||||
And a Log created
|
||||
When the log method is called
|
||||
Then this new log must be registered in the Replay Data Base
|
||||
|
||||
|
||||
# Since this is a remote repository, it is impossible to test the config getters methods.
|
||||
# They have to be runned inside a local repository wich contains Replay, so, unfortunatelly, they will not be tested, at least not for now.
|
||||
|
||||
|
||||
#It is also not possible to test the open_app method for the same reasons as above.
|
||||
|
||||
|
||||
Scenario: Testing the menu_get_all_enabled method.
|
||||
Given the Replay client
|
||||
When the menu_get_all_enabled method is called
|
||||
Then there must be a list as a return
|
||||
|
||||
|
||||
# Given the fact that the methods "service_stop_all" and "exit" will break the tests chain, they cannot be tested.
|
||||
|
||||
|
||||
# The same from above goes for the "exit" method.
|
||||
|
||||
|
||||
# The queue methods needs some parameters, wich cannot be given, since this test is not made to be launched through Replay and there is no way to know wich are the "job_id"s or "queue_id"s from the tester machine.
|
||||
|
||||
|
||||
Scenario: Testing queue_abort method
|
||||
Given the Replay client
|
||||
When the queue_abort method is called
|
||||
Then there must be None as return, since this test was not runned by Replay
|
|
@ -0,0 +1,79 @@
|
|||
from behave import *
|
||||
from cli import Replay
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
# ============================= ENVIRONMENT VARIABLES GETTERS =============================
|
||||
@given(u'the Replay client')
|
||||
def step_impl (context):
|
||||
context.client = Replay ()
|
||||
|
||||
@when(u'each one of the getters is called')
|
||||
def step_impl(context):
|
||||
context.returns = {context.client.replay_env_queue_id(): "context.client.replay_env_queue_id()", context.client.replay_env_alias(): "context.client.replay_env_alias()", context.client.replay_env_feature_dir(): "context.client.replay_env_feature_dir()", context.client.replay_env_root(): "context.client.replay_env_root()", context.client.replay_env_addr(): "context.client.replay_env_addr()", context.client.replay_env_home_dir(): "context.client.replay_env_home_dir()", context.client.replay_env_repo(): "context.client.replay_env_repo()", context.client.replay_env_ver(): "context.client.replay_env_ver()", context.client.replay_env_repo_dir(): "context.client.replay_env_repo_dir()", context.client.replay_env_data_dir(): "context.client.replay_env_data_dir()", context.client.replay_env_instance_alias(): "context.client.replay_env_instance_alias()", context.client.replay_env_api_key(): "context.client.replay_env_api_key()"}
|
||||
|
||||
@then(u'all of the returns must be either of the type string or the type None')
|
||||
def step_impl(context):
|
||||
for ret in context.returns.keys():
|
||||
assert ( type(ret) == type(None) or type(ret) == type("") ), f'Something went wrong with the {context.returns[ret]} variable getter.'
|
||||
|
||||
|
||||
|
||||
# ============================= LOG REGISTERS =============================
|
||||
@given(u'a Log created')
|
||||
def step_impl(context):
|
||||
context.log = context.client.new_log()
|
||||
|
||||
@when(u'the log method is called')
|
||||
def step_impl(context):
|
||||
context.dt_log_end = datetime.now().isoformat() + "Z"
|
||||
context.log_return = context.client.log(context.log)
|
||||
|
||||
@then(u'this new log must be registered in the Replay Data Base')
|
||||
def step_impl(context):
|
||||
assert (context.client.sql("select * from logs")["data"][-1]["dtlogend"] == context.dt_log_end and context.log_return == "ok")
|
||||
|
||||
|
||||
|
||||
# ============================= CONFIGURATION GETTERS =============================
|
||||
# Since this is a remote repository, it is impossible to test the config getters methods.
|
||||
# They have to be runned inside a local repository wich contains Replay, so, unfortunatelly, they will not be tested, at least not for now.
|
||||
|
||||
|
||||
|
||||
# ============================= OPEN APP =============================
|
||||
#It is also not possible to test the open_app method for the same reasons as above.
|
||||
|
||||
|
||||
|
||||
# ============================= GET ALL ENABLED MENUS =============================
|
||||
@when(u'the menu_get_all_enabled method is called')
|
||||
def step_impl(context):
|
||||
context.all_menus_return = context.client.menu_get_all_enabled()
|
||||
|
||||
@then(u'there must be a list as a return')
|
||||
def step_impl(context):
|
||||
assert type(context.all_menus_return) == type([]), "Something went wrong getting the enabled menus."
|
||||
|
||||
|
||||
|
||||
# Given the fact that the methods "service_stop_all" and "exit" will break the tests chain, they cannot be tested.
|
||||
|
||||
|
||||
|
||||
# The same from above goes for the "exit" method.
|
||||
|
||||
|
||||
|
||||
# The queue methods needs some parameters, wich cannot be given, since this test is not made to be launched through Replay and there is no way to know wich are the "job_id"s or "queue_id"s from the tester machine.
|
||||
|
||||
|
||||
|
||||
# ============================= QUEUE ABORT =============================
|
||||
@when(u'the queue_abort method is called')
|
||||
def step_impl(context):
|
||||
context.queue_abort_return = context.client.queue_abort()
|
||||
|
||||
@then(u'there must be None as return, since this test was not runned by Replay')
|
||||
def step_impl(context):
|
||||
assert (context.queue_abort_return == None), 'Something went wrong with queue_abort.'
|
|
@ -17,7 +17,7 @@ class Wingui:
|
|||
ep: str = ""
|
||||
|
||||
def __init__(self):
|
||||
self.ep = "https://localhost:8443"
|
||||
self.ep = os.environ.get("REPLAY_ADDR")
|
||||
|
||||
def __requestget__(self, data: str):
|
||||
|
||||
|
|
|
@ -0,0 +1,205 @@
|
|||
from time import sleep
|
||||
from behave import *
|
||||
from cli import Wingui
|
||||
|
||||
# =========================== Clip_read & Clip_write ===========================
|
||||
@given(u'a winGUI client')
|
||||
def step_impl(context):
|
||||
context.client = Wingui ()
|
||||
|
||||
@when(u'the clip_write method is called to store the text "There is a light that never goes out"')
|
||||
def step_impl(context):
|
||||
context.client.clip_write("There is a light that never goes out")
|
||||
|
||||
@then(u'when the clip_read method is called, it must return the same text')
|
||||
def step_impl(context):
|
||||
assert context.client.clip_read() == "There is a light that never goes out", "Something went wrong with the clipboard."
|
||||
|
||||
|
||||
|
||||
# =========================== Screen_scale & Screen_size ===========================
|
||||
@when(u'the screen_scale method is called')
|
||||
def step_impl(context):
|
||||
context.scale = context.client.screen_scale()
|
||||
|
||||
@when(u'the screen_size method is also called')
|
||||
def step_impl(context):
|
||||
context.size = context.client.screen_size()
|
||||
|
||||
@then(u'the returns of both must be equal')
|
||||
def step_impl(context):
|
||||
assert context.scale == context.size, "Something went wrong while getting the screen dimensions."
|
||||
|
||||
# Please note that most of the other screen methods use image recognition and, because of that, it is not possible to test them on every computer, since there is no way to know wich images are being displayed on the screen.
|
||||
# Also, the other methods wich do not use image recognition are simply too difficult to test.
|
||||
|
||||
|
||||
|
||||
# =========================== Proc_exec & Proc_all ===========================
|
||||
@when(u'the proc_exec method is called with "mspaint"')
|
||||
def step_impl(context):
|
||||
context.client.proc_exec("mspaint")
|
||||
|
||||
@when(u'the proc_all method is called')
|
||||
def step_impl(context):
|
||||
context.procs = context.client.proc_all()
|
||||
|
||||
@then(u'there must be a process with the name "mspaint.exe" inside the list')
|
||||
def step_impl(context):
|
||||
for proc in context.procs:
|
||||
if "mspaint.exe" in proc["Name"]:
|
||||
context.client.proc_kill(proc["Pid"])
|
||||
assert True
|
||||
return
|
||||
|
||||
assert False, "Something went wrong executing Microsoft Paint."
|
||||
|
||||
|
||||
|
||||
# =========================== Proc_name ===========================
|
||||
@given(u'the initialization of Microsoft Paint')
|
||||
def step_impl(context):
|
||||
context.client.proc_exec("mspaint")
|
||||
|
||||
@given(u'the "Name" and "Pid" infos about this process')
|
||||
def step_impl(context):
|
||||
context.procs = context.client.proc_all()
|
||||
|
||||
for proc in context.procs:
|
||||
if "mspaint.exe" in proc["Name"]:
|
||||
context.proc_name = proc["Name"]
|
||||
context.proc_pid = proc["Pid"]
|
||||
|
||||
@when(u'the proc_name method is called with the "Pid" collected')
|
||||
def step_impl(context):
|
||||
context.name_return = context.client.proc_name(context.proc_pid)
|
||||
|
||||
@then(u'the return must be te same as the "Name" collected')
|
||||
def step_impl(context):
|
||||
assert context.name_return == context.proc_name, "Something went wrong with some of the proc methods."
|
||||
|
||||
context.client.proc_kill(context.proc_pid)
|
||||
|
||||
|
||||
|
||||
# =========================== Proc_pids ===========================
|
||||
@when(u'the proc_pids is called')
|
||||
def step_impl(context):
|
||||
context.all_pids = context.client.proc_pids()
|
||||
|
||||
@then(u'the "Pid" info collected must be inside the array returned')
|
||||
def step_impl(context):
|
||||
assert context.proc_pid in context.all_pids, "Something went wrong catching all of the pids."
|
||||
|
||||
context.client.proc_kill(context.proc_pid)
|
||||
|
||||
|
||||
|
||||
# =========================== Window_list and Window_hwnd ===========================
|
||||
@when(u'the window_hwnd method is called with the parameter "mspaint.exe"')
|
||||
def step_impl(context):
|
||||
sleep(0.5)
|
||||
context.paint_hwnd = context.client.window_hwnd("mspaint.exe")
|
||||
|
||||
@when(u'the window_list method is called')
|
||||
def step_impl(context):
|
||||
context.windows = context.client.window_list()
|
||||
|
||||
@then(u'the return of window_hwnd must be the value of one of the "Hwnd" keys returned in window_list')
|
||||
def step_impl(context):
|
||||
for window in context.windows:
|
||||
if context.paint_hwnd[0] == window["Hwnd"]:
|
||||
assert True
|
||||
sleep(0.5)
|
||||
context.client.proc_kill(context.proc_pid)
|
||||
|
||||
return
|
||||
|
||||
assert False, "Something went wrong catching the Paint window."
|
||||
|
||||
|
||||
|
||||
# # =================== Window_activate, Window_activetitle and Window_activehwnd ===================
|
||||
# @given(u'the window hwnd obtained from the "Name" info')
|
||||
# def step_impl(context):
|
||||
# context.paint_hwnd = context.client.window_hwnd(context.proc_name)
|
||||
|
||||
# @when(u'the window_activate method is called with the Paint Hwnd obtained')
|
||||
# def step_impl(context):
|
||||
# sleep(0.5)
|
||||
# context.client.window_activate (context.paint_hwnd[0])
|
||||
# sleep(0.5)
|
||||
|
||||
# @when(u'the methods window_activetitle and window_activehwnd are called')
|
||||
# def step_impl(context):
|
||||
# context.paint_title = context.client.window_activetitle()
|
||||
# context.paint_win_hwnd = context.client.window_activehwnd()
|
||||
|
||||
# context.client.window_close(context.paint_hwnd[0])
|
||||
|
||||
# @then(u'the returned value from window_activetitle must contain the substring "Paint"')
|
||||
# def step_impl(context):
|
||||
# assert "Paint" in context.paint_title, f"The active title obtained is not from Paint.\nActive Title: {context.paint_title}"
|
||||
|
||||
# for window in context.client.window_list():
|
||||
# if context.paint_hwnd == window["Hwnd"]:
|
||||
# print(f"""
|
||||
|
||||
# ---------------------------
|
||||
# Hwnd associated title: {window["Title"]}
|
||||
# ---------------------------
|
||||
|
||||
# """)
|
||||
|
||||
# @then(u'the returned value from window_activehwnd must be equal to the hwnd previously obtained')
|
||||
# def step_impl(context):
|
||||
# assert context.paint_win_hwnd == context.paint_hwnd, "The Hwnd obtained is not from Paint"
|
||||
|
||||
|
||||
|
||||
# ======================================== Display methods ========================================
|
||||
@given(u'the current resolution of the display')
|
||||
def step_impl(context):
|
||||
context.sys_info = context.client.display_res()
|
||||
|
||||
context.prev_width = context.sys_info["DmPelsWidth"]
|
||||
context.prev_height = context.sys_info["DmPelsHeight"]
|
||||
|
||||
@when(u'the display_setres method is called to set the display resolution in 800x600')
|
||||
def step_impl(context):
|
||||
context.new_sys_info = context.sys_info
|
||||
context.new_sys_info["DmPelsWidth"] = 800
|
||||
context.new_sys_info["DmPelsHeight"] = 600
|
||||
|
||||
try:
|
||||
context.client.display_setres(context.new_sys_info)
|
||||
sleep(5)
|
||||
except:
|
||||
raise "Your system configuration does not support 800x600 resolution"
|
||||
|
||||
@then(u'the new display resolution must be 800x600')
|
||||
def step_impl(context):
|
||||
context.new_sys_info = context.client.display_res()
|
||||
|
||||
assert context.new_sys_info["DmPelsWidth"] == 800 and \
|
||||
context.new_sys_info["DmPelsHeight"] == 600, \
|
||||
\
|
||||
"Something went wrong setting your system resolution to 800x600.\n\nCheck if this resolution is available in your configurations, if so, please disregard this message."
|
||||
|
||||
@then(u'the methods must be able to set your resolution to the previous one')
|
||||
def step_impl(context):
|
||||
context.new_sys_info["DmPelsWidth"] = context.prev_width
|
||||
context.new_sys_info["DmPelsHeight"] = context.prev_height
|
||||
|
||||
try:
|
||||
context.client.display_setres(context.new_sys_info)
|
||||
sleep(5)
|
||||
except:
|
||||
raise "Error!"
|
||||
|
||||
context.curr_sys_info = context.client.display_res()
|
||||
|
||||
assert context.curr_sys_info["DmPelsWidth"] == context.prev_width and \
|
||||
context.curr_sys_info["DmPelsHeight"] == context.prev_height, \
|
||||
\
|
||||
"Sorry for the inconvenience, but something went wrong resetting your resolution to the previous one.\n\nPlease go to your system configurations to do this step manually."
|
|
@ -0,0 +1,73 @@
|
|||
Feature: The winGUI API client
|
||||
|
||||
Scenario: Testing the clip_read and clip_write methods
|
||||
Given a winGUI client
|
||||
When the clip_write method is called to store the text "There is a light that never goes out"
|
||||
Then when the clip_read method is called, it must return the same text
|
||||
|
||||
|
||||
Scenario: Testing the screen_scale and screen_size methods
|
||||
Given a winGUI client
|
||||
When the screen_scale method is called
|
||||
And the screen_size method is also called
|
||||
Then the returns of both must be equal
|
||||
|
||||
# Please note that most of the other screen methods use image recognition and, because of that, it is not possible to test them on every computer, since there is no way to know wich images are being displayed on the screen.
|
||||
# Also, the other methods wich do not use image recognition are simply too difficult to test.
|
||||
|
||||
|
||||
Scenario: Testing proc_exec and proc_all methods
|
||||
Given a winGUI client
|
||||
When the proc_exec method is called with "mspaint"
|
||||
And the proc_all method is called
|
||||
Then there must be a process with the name "mspaint.exe" inside the list
|
||||
|
||||
|
||||
Scenario: Testing proc_name method
|
||||
Given a winGUI client
|
||||
And the initialization of Microsoft Paint
|
||||
And the "Name" and "Pid" infos about this process
|
||||
When the proc_name method is called with the "Pid" collected
|
||||
Then the return must be te same as the "Name" collected
|
||||
|
||||
|
||||
Scenario: Testing proc_pids method
|
||||
Given a winGUI client
|
||||
And the initialization of Microsoft Paint
|
||||
And the "Name" and "Pid" infos about this process
|
||||
When the proc_pids is called
|
||||
Then the "Pid" info collected must be inside the array returned
|
||||
|
||||
|
||||
Scenario: Testing the window_list and window_hwnd methods
|
||||
Given a winGUI client
|
||||
And the initialization of Microsoft Paint
|
||||
And the "Name" and "Pid" infos about this process
|
||||
|
||||
When the window_hwnd method is called with the parameter "mspaint.exe"
|
||||
And the window_list method is called
|
||||
|
||||
Then the return of window_hwnd must be the value of one of the "Hwnd" keys returned in window_list
|
||||
|
||||
|
||||
# Scenario: Testing window_activate, window_activetitle and window_activehwnd
|
||||
# Given a winGUI client
|
||||
# And the initialization of Microsoft Paint
|
||||
# And the "Name" and "Pid" infos about this process
|
||||
# And the window hwnd obtained from the "Name" info
|
||||
|
||||
# When the window_activate method is called with the Paint Hwnd obtained
|
||||
# And the methods window_activetitle and window_activehwnd are called
|
||||
|
||||
# Then the returned value from window_activetitle must contain the substring "Paint"
|
||||
# And the returned value from window_activehwnd must be equal to the hwnd previously obtained
|
||||
|
||||
|
||||
Scenario: Testing the display methods
|
||||
Given a winGUI client
|
||||
And the current resolution of the display
|
||||
|
||||
When the display_setres method is called to set the display resolution in 800x600
|
||||
|
||||
Then the new display resolution must be 800x600
|
||||
And the methods must be able to set your resolution to the previous one
|
Loading…
Reference in New Issue