Tests for Mail and Wpp, Mail works

Luciano Barletta
Commit a0f76e52 authored 2019-10-03 15:54:21 +0000 by Luciano Barletta
Showing 6 changed files with 128 additions and 49 deletions
database.py
deploy.py
enums.py
process.py
services.py
test_flask.py
--- a/database.py
View file @a0f76e5
+++ b/database.py
View file @a0f76e5
-import sqlite3
-import ipdb
 from enums import Table, States, Services, Datatypes
+import sqlite3, ipdb, json
 class DBconnection:
@@ -76,7 +75,6 @@ class DBconnection:
        return "El dato '" + alterations[column] + "' no es valido"
      query += column + "='" + alterations[column] + "',"
    query = query.strip(",")
-    ipdb.set_trace()
    con = sqlite3.connect(self.db)
    cursor = con.cursor()
    cursor.execute(query + where)
@@ -91,7 +89,11 @@ class DBconnection:
    if column == Table.dest:
      return True
    if column == Table.type:
-      return Datatypes.validate(data)
+      data = json.loads(data)
+      valid = True
+      for key in data:
+        valid = valid and Datatypes.validate(data[key])
+      return valid
    if column == Table.state:
      return States.validate(data)
    return False

--- a/deploy.py
View file @a0f76e5
+++ b/deploy.py
View file @a0f76e5
 #-*- coding: utf-8 -*-
 from flask import Flask, render_template, json, request, url_for
 from werkzeug.datastructures import FileStorage
-import os
-import ipdb
 from process import Process
-import time, threading
 from python_arptable import get_arp_table
-import random
 from enums import States, Table
-import datetime
+import os, ipdb, time, threading, random, datetime
 app = Flask(__name__)
 retry_timer = 10
@@ -89,12 +85,12 @@ def msg():
 @app.route('/cons', methods = ['POST'])
 def cons():
-  id_query = request.form['id']
+  id_query = request.values['id']
  row = process.lookup(id_query)
  if type(row) == str:  # error message
    return row 
  if row[Table.state] == States.delivered:
-    os.system("rm -r" + row[Table.path])
+    os.system("rm -r " + row[Table.path])
  return str(row[Table.state])
 def attempt():

--- a/enums.py
View file @a0f76e5
+++ b/enums.py
View file @a0f76e5
@@ -21,7 +21,7 @@ class States:
 class Services:
  wpp1 = "wpp1"
-  sms = "sms"
+  mail = "mail"
  @staticmethod
  def validate(serv):
@@ -33,6 +33,7 @@ class Datatypes:
  document = "document"
  link = "link"
  audio = "audio"
+  html = "html"
  @staticmethod
  def validate(datatype):

--- a/process.py
View file @a0f76e5
+++ b/process.py
View file @a0f76e5
-import ipdb
-import os
-import json
 from services import serviceFactory
 from database import DBconnection
 from enums import Services, States, Datatypes, Table
+import ipdb, os, json
 class Process:
@@ -25,7 +23,6 @@ class Process:
    # service is wrong
    if not Services.validate(query[Table.serv]):
      return "No existe el servicio '" + query[Table.serv] + "'"
-    ipdb.set_trace()
    types = json.loads(query[Table.type])
    filelist = os.listdir(
      self.lookup(
@@ -35,10 +32,10 @@ class Process:
    for file in types:
      # files don't exist
      if file not in filelist:
-        return "El archivo '" + file "' no existe"
+        return "El archivo '" + file + "' no existe"
      # message can't be sent by this service
      elif not serviceFactory(query[Table.serv]).validate(types[file]):
-        return "El servicio '" + query[Table.serv] + "' no puede enviar el tipo '" + types[file] + "' destinado al archivo '" file "'" 
+        return "El servicio '" + query[Table.serv] + "' no puede enviar el tipo '" + types[file] + "' destinado al archivo '" + file + "'" 
    entities = {
      Table.dest : query[Table.dest],
@@ -46,7 +43,9 @@ class Process:
      Table.type : query[Table.type],
      Table.state : States.queued
    }
-    self.conn.update("msg",(Table.id,query[Table.id]),entities)
+    error = self.conn.update("msg",(Table.id,query[Table.id]),entities)
+    if error:
+      return error
    return States.queued
  # tries to send all messages available

--- a/services.py
View file @a0f76e5
+++ b/services.py
View file @a0f76e5
-import requests
-import json
-import os
 from enums import Table, Services, States, Datatypes
 from abc import ABC, abstractmethod
-import smtplib
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 from email.mime.image import MIMEImage
 from email.mime.audio import MIMEAudio
+from email.mime.application import MIMEApplication
+import requests, json, os, smtplib
 class ServiceBase(ABC):
@@ -34,6 +32,7 @@ class Wpp1(ServiceBase):
  def send(self,data):
    types = json.loads(data[Table.type])
+    succ = True
    for file in types:
      filepath = data[Table.path] + file
      if types[file] == Datatypes.text:
@@ -41,17 +40,18 @@ class Wpp1(ServiceBase):
        text = f.read()
        f.close()
        result = requests.get(url = Wpp1.URL + Wpp1.URLmode[types[file]],params = {'token':Wpp1.token,'uid':Wpp1.uid,'to':data[Table.dest],'text':text})
-        return result.json()['success']
+        succ = succ and result.json()['success']
      else:
        path = requests.post(url = Wpp1.server, files = { 'data' : (file,open(filepath,'rb')) })
-        result = requests.get(url = Wpp1.URL + Wpp1.URLmode[data[Table.type]],params = {'token':Wpp1.token,'uid':Wpp1.uid,'to':data[Table.dest],'url':Wpp1.server + path.text})
+        result = requests.get(url = Wpp1.URL + Wpp1.URLmode[types[file]],params = {'token':Wpp1.token,'uid':Wpp1.uid,'to':data[Table.dest],'url':Wpp1.server + path.text})
-        return result.json()['success']
+        succ = succ and result.json()['success']
+    return succ
  def validate(self,datatype):
    return datatype == Datatypes.text or datatype == Datatypes.image or datatype == Datatypes.document or datatype == Datatypes.link
-class SMS(ServiceBase):
+class Mail(ServiceBase):
  def __init__(self):
    self.__username = "prueba@anacsoft.com"
@@ -62,28 +62,36 @@ class SMS(ServiceBase):
  def send(self,data):
    msg = MIMEMultipart()
-    msg['From'] = self._SMS__username
+    msg['From'] = self._Mail__username
    msg['To'] = data[Table.dest]
    msg['Subject'] = "Test"
    types = json.loads(data[Table.type])
    for file in types:
      filepath = data[Table.path] + file
-      msg.attach(self.MIMEmode(filepath, types[file]))
+      MIME = self.MIMEmode(filepath, types[file])
+      msg.attach(MIME)
    self.s.send_message(msg)
    return True
  def MIMEmode(self,path,type):    
-    data = open(path,'rb')    
    mode = None
    if type == Datatypes.text:
+      with open(path) as data:
        mode = MIMEText(data.read(),'plain')
+    if type == Datatypes.html:
+      with open(path) as data:
+        mode = MIMEText(data.read(),'html')
    if type == Datatypes.image:
+      with open(path, "rb") as data:
        mode = MIMEImage(data.read())
    if type == Datatypes.audio:
+      with open(path, "rb") as data:
        mode = MIMEAudio(data.read())
+    if type == Datatypes.document:
+      with open(path, "rb") as data:
+        mode = MIMEApplication(data.read())
-    f.close()
    return mode
  def validate(self,datatype):
@@ -93,11 +101,15 @@ class SMS(ServiceBase):
      return True
    if datatype == Datatypes.audio:
      return True
+    if datatype == Datatypes.document:
+      return True
+    if datatype == Datatypes.html:
+      return True
    return False
 def serviceFactory(serv):
  if serv == Services.wpp1:
    return Wpp1()
-  if serv == Services.sms:
+  if serv == Services.mail:
-    return SMS()
+    return Mail()
  return None
--- a/test_flask.py
View file @a0f76e5
+++ b/test_flask.py
View file @a0f76e5
-from flask import Flask, render_template
+import os, time, requests, ipdb, json, threading
-from flask_testing import TestCase
-import unittest
-class RootTest(TestCase):
+URL = "http://192.168.15.75:5000/"
+TESTPHONE = "5493415959169"
+TESTMAIL = "sujetodeprueba0110@gmail.com"
+FOLDER = "testfolder/"
+SENDING_TIME = 40 # aumentar a medida que se vuelva necesario
-  render_templates = True
+def wpp1(phone):
+  id = requests.post(url = URL + "data", files = {
+    'wpmsg' : open(FOLDER + "wptext","rb"),
+    'wpimage' : open(FOLDER + "wpimage","rb"),
+    'wpmedia' : open(FOLDER + "wpmedia","rb"),
+    'wplink' : open(FOLDER + "wplink","rb")
+  })
+  assert int(id.text) > 0 , id.text + "no es mayor a 0"
+  state = requests.post(url = URL + "msg", params = {
+    'id' : id.text,
+    'serv' : "wpp1",
+    'dest' : phone,
+    'type' : json.dumps({
+      'wptext' : 'text',
+      'wpimage' : 'image',
+      'wpmedia' : 'document',
+      'wplink' : 'link'
+    })
+  })
+  assert state.text == "queued" , "'" + state.text + "' no es igual a 'queued'"
+  time.sleep(SENDING_TIME) 
+  state = requests.post(url = URL + "cons", params = {
+    'id' : id.text
+  })
+  assert (state.text == "delivered") , "'" + state.text + "' no es igual a 'delivered'"
-  def create_app(self):
+def mail(mail):
-    app = Flask(__name__)
+  id = requests.post(url = URL + "data", files = {
-    app.config['Testing'] = True
+    'mailmsg' : open(FOLDER + "mailtext","rb"),
+    'mailimage' : open(FOLDER + "mailimage","rb"),
+    'maildocument' : open(FOLDER + "maildocument","rb")
+  })
+  assert int(id.text) > 0 , id.text + "no es mayor a 0"
+  state = requests.post(url = URL + "msg", params = {
+    'id' : id.text,
+    'serv' : "mail",
+    'dest' : mail,
+    'type' : json.dumps({
+      'mailtext' : 'text',
+      'mailimage' : 'image',
+      'maildocument' : 'document'
+    })
+  })
+  assert state.text == "queued" , "'" + state.text + "' no es igual a 'queued'"
+  time.sleep(SENDING_TIME) 
+  state = requests.post(url = URL + "cons", params = {
+    'id' : id.text
+  })
+  assert (state.text == "delivered") , "'" + state.text + "' no es igual a 'delivered'"
-    @app.route('/')
+def errors():
-    def main():
+  # service not valid
-      return render_template('index.html')
+  # service unable to send type
+  # trying to set type of unexistent file
+  # trying to user unexistent table
+  # ill-formed comparator
+  # data doesn't pass the checks
-    return app
+def main():
+  if os.path.exists("messages.db"):
+    os.remove("messages.db")
-  def test_something(self):
+  try:
-    response = self.client.get("/")
+    os.system("python3 deploy.py &")
-    self.assertEqual(response.data.decode('utf-8'),"Nothing to see here...",response)
+    time.sleep(3)
+    threads = []
+    threads.append(threading.Thread(target = wpp1, args = (TESTPHONE,)))
+    threads.append(threading.Thread(target = encryption, args = (TESTPHONE,)))
+    threads.append(threading.Thread(target = mail, args = (TESTMAIL,)))
+    for thread in threads:
+      thread.start()
+    for thread in threads:
+      thread.join()
+  except Exception as exp:
+    print(type(exp))
+    print(exp.args)
+  finally:
+    os.system("killall python3")
 if __name__ == "__main__":
-  unittest.main()
\ No newline at end of file
+  main()
\ No newline at end of file