如何同步Python脚本与Espresso测试，实现Firebase实时数据库E2E测试？

针对你的需求——不模拟Firebase、依赖真实模拟器延迟、同步Python数据库操作与Espresso测试，以下是几个直接落地的方法：

1. 本地HTTP/Socket通信实现实时同步 #

这是最灵活的方案，通过跨进程通信让Python和Espresso互相触发执行：

• Espresso端：在测试代码中启动轻量HTTP服务器，监听模拟器内部端口（如8080）。当测试需要等待Python的数据库更新时，阻塞等待服务器收到确认请求。

  // Espresso测试中启动HTTP服务器并等待信号
  private val syncLatch = CountDownLatch(1)
  
  @Before
  fun setupSyncServer() {
      val server = com.sun.net.httpserver.HttpServer.create(InetSocketAddress(8080), 0)
      server.createContext("/update-done") { exchange ->
          syncLatch.countDown()
          exchange.sendResponseHeaders(200, 0)
          exchange.close()
      }
      server.start()
  }
  
  @Test
  fun testFirstDataUpdate() {
      // 等待Python完成数据库更新
      syncLatch.await(30, TimeUnit.SECONDS)
      // 执行UI验证
      onView(withId(R.id.data_text)).check(matches(withText("更新后的数据")))
  }
  

• Python端：完成Realtime Database更新后，向模拟器的映射端口发送HTTP请求（模拟器访问主机用10.0.2.2）：

  import requests
  import firebase_admin
  from firebase_admin import db
  
  # 初始化Firebase Admin连接模拟器
  cred = firebase_admin.credentials.ApplicationDefault()
  firebase_admin.initialize_app(cred, {
      'databaseURL': 'http://localhost:9000/?ns=your-project-id'
  })
  
  # 更新数据库
  ref = db.reference('test/data')
  ref.set({'value': '更新后的数据'})
  
  # 通知Espresso测试继续
  requests.get('http://10.0.2.2:8080/update-done')
  

2. 利用Firebase模拟器事件监听确保操作完成 #

通过监听Firebase模拟器的数据库写入事件，确保Python的更新确实生效后再触发测试：

• Python端：在更新数据后，监听目标节点的写入事件，确认数据同步完成：

  import threading
  
  def wait_for_update(ref, expected_value):
      callback_done = threading.Event()
      def callback(event):
          if event.event_type == 'put' and event.data == expected_value:
              callback_done.set()
      ref.listen(callback)
      callback_done.wait(timeout=30)
  

• Espresso端：结合IdlingResource监听Firebase数据库状态，当数据更新后自动解除Espresso的等待，无需硬编码延迟：

  class FirebaseIdlingResource(private val ref: DatabaseReference, private val expectedValue: String) : IdlingResource {
      private var callback: ResourceCallback? = null
      private var isIdle = false
  
      init {
          ref.addValueEventListener(object : ValueEventListener {
              override fun onDataChange(snapshot: DataSnapshot) {
                  isIdle = snapshot.getValue(String::class.java) == expectedValue
                  callback?.onTransitionToIdle()
              }
              override fun onCancelled(error: DatabaseError) {}
          })
      }
  
      override fun getName() = "FirebaseIdlingResource"
      override fun isIdleNow() = isIdle
      override fun registerIdleTransitionCallback(callback: ResourceCallback?) {
          this.callback = callback
      }
  }
  
  // 在测试中注册IdlingResource
  @Test
  fun testDataUpdate() {
      val idlingResource = FirebaseIdlingResource(ref, "更新后的数据")
      IdlingRegistry.getInstance().register(idlingResource)
      onView(withId(R.id.data_text)).check(matches(withText("更新后的数据")))
      IdlingRegistry.getInstance().unregister(idlingResource)
  }
  

3. 用Shell脚本串联测试流程 #

如果不需要实时双向同步，只是按顺序执行步骤，用Shell脚本统一调度Python脚本和Espresso测试命令最简单：

#!/bin/bash

# 第一步：执行Python脚本更新数据库
python update_db_step1.py

# 第二步：运行对应的Espresso测试用例
adb shell am instrument -w -e class com.your.app.TestClass#testFirstUpdate com.your.app.test/androidx.test.runner.AndroidJUnitRunner

# 第三步：执行第二次数据库更新
python update_db_step2.py

# 第四步：运行第二个测试用例
adb shell am instrument -w -e class com.your.app.TestClass#testSecondUpdate com.your.app.test/androidx.test.runner.AndroidJUnitRunner

这种方式适合线性执行的测试流程，无需额外通信逻辑，缺点是无法处理异步延迟，需要确保Python脚本和测试的执行时间足够覆盖数据库同步延迟。

关键注意事项 #

• 模拟器网络映射：Android模拟器访问主机服务时必须用10.0.2.2，不要用localhost。
• 测试隔离：每个测试前通过Firebase模拟器的REST API重置数据库（curl -X DELETE http://localhost:9000/your-project-id.json），避免测试数据污染。
• 避免硬编码等待：尽量用事件驱动的等待（如IdlingResource、Latch）替代Thread.sleep()，保证测试稳定性。

内容的提问来源于stack exchange，提问作者Mr T