1

In Appium (using Python) I want to send the app to the background, do some work while the app is in background (that would send a push notification to the app), and then return to the app.

I tried to do it with threads (threading.Thread Python class), but it wouldn't do it in parallel - the app would sit in background, and after that was done, the second thread would start.

Then I tried it with multiprocessing.Process, but I'm getting AttributeError: Can't pickle local object 'PoolManager.__init__.<locals>.<lambda>'

How should I go about this?

These are the parts of the code I have now (as I understand, the functions that should be parallelized need to be outside of the class):

def send_app_to_background(app_driver, seconds=10):
    app_driver.background_app(seconds)

def asdf_verify_user(page:OnboardingPage, passenger_email):
    mailer = MyMailer(passenger_email, page.settings)
    verification_url = mailer.get_verification_url()
    page.open_verification_url(verification_url)

And this is how it's called from inside the test class:

class TestMails(kh.BasicTest):
    def test_registration_mails(self):
        ........
        jobs = []
        process1 = Process(target=send_app_to_background, args=(self.app_driver, 20))
        process2 = Process(target=verify_user, args=(page, passenger_email))
        jobs.append(process1)
        jobs.append(process2)
        for j in jobs:
            j.start()
        
        for j in jobs:
            j.join()
zorglub76
  • 4,852
  • 7
  • 37
  • 46
  • Use a thread to run your function. See this answer: https://stackoverflow.com/questions/71601983/runtime-error-when-trying-to-spawn-multiple-processes-image-detection-and-print/71602709#71602709 – Cow Jun 01 '22 at 08:32
  • I tried it with asyncio. Those functions stayed the same (except for the 'async' keyword in front), and I created another function to call them: async def asdf(app_driver, page:OnboardingPage, passenger_email): task1 = asyncio.create_task(send_app_to_background(app_driver, 20)) task2 = asyncio.create_task(verify_user(page, passenger_email)) await task1 await task2 and then I call that from the test like asyncio.run(asdf(self.app_driver, page, passenger_email)) But it's still sequential...... – zorglub76 Jun 01 '22 at 09:50
  • I think the problem is that `background_app(seconds)`, which is a function of the Appium object, is not async.. I tried playing with two async functions that use `time.sleep()` instead of `await asyncio.sleep()`, and they behave exactly as my functions above.... – zorglub76 Jun 01 '22 at 10:55

1 Answers1

0

Well, apparently, it's not possible to do it like this. Appium's background_app() function is locking the other threads. My solution was to use close_app() and launch_app() instead.

zorglub76
  • 4,852
  • 7
  • 37
  • 46